Shulou(Shulou.com)06/02 Report--

This article mainly introduces "flinksql how to link kafka". In daily operation, I believe many people have doubts about how flinksql links to kafka. The editor consulted all kinds of materials and sorted out simple and easy-to-use methods of operation. I hope it will be helpful to answer the doubts of "flinksql how to link kafka"! Next, please follow the editor to study!

4.0.0 org.example flinksqldemo 1.0-SNAPSHOT UTF-8 UTF-8 2.11 2.11.8 0.10.2.1 1.12.0 2.7.3 compile org.apache.maven.plugins maven- Compiler-plugin 8 8 org.apache.flink flink-table-planner-blink_2.11 1.12.0 org.apache.flink Flink-java ${flink.version} ${setting.scope} org.apache.flink flink-streaming-java_2.11 ${flink.version} ${setting.scope} org.apache.flink flink-clients_2.11 ${ Flink.version} ${setting.scope} org.apache.flink flink-connector-kafka_2.11 1.12.0 org.apache.flink flink-csv 1.12.0 org.apache.flink flink- Streaming-scala_$ {scala.binary.version} ${flink.version} ${setting.scope} org.apache.kafka kafka_$ {scala.binary.version} ${kafka.version} ${setting.scope} Org.apache.hadoop hadoop-common ${hadoop.version} ${setting.scope} org.apache.hadoop hadoop-hdfs ${hadoop.version} ${setting.scope} org.apache.hadoop hadoop- Client ${hadoop.version} ${setting.scope} org.slf4j slf4j-api 1.7.25 com.alibaba fastjson 1.2.72 redis.clients jedis 2.7.3 com.google.guava guava 29.0-jre

Code:

Package com.jd.data;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.DataTypes;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import org.apache.flink.table.descriptors.Csv;import org.apache.flink.table.descriptors.Kafka;import org.apache.flink.table.descriptors.Schema;import org.apache.flink.types.Row Public class TableApiConnectKafka04 {public static void main (String [] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); env.setParallelism (1); / / 1. Create table execution StreamTableEnvironment tableEnv = StreamTableEnvironment.create (env) TableEnv.connect (new Kafka () .version ("0.11") / / define the version .topic ("xxx") / / define the topic .property ("zookeeper.connect", "localhost:2181") .property ("bootstrap.servers") "localhost:9092") .withFormat (new Csv ()) .withSchema (new Schema (). Field ("a", DataTypes.STRING ()) / / defines the structure of the table .field ("b", DataTypes.STRING ()) .field ("c", DataTypes.STRING ()) .inAppendMode () .createTemporaryTable ("xxx") Table xxx = tableEnv.from ("xxx"); xxx.printSchema (); tableEnv.toAppendStream (xxx, Row.class). Print (); env.execute ("job");}} at this point, the study of "how flinksql links kafka" is over, hoping to solve everyone's doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.