FlinkSQL kafka完整案例 可直接复制使用

为自己记录一下flinksql 消费kafka json数据 并写入doris的完整案例
用完发现，flinksql 是真的香。

虽然尽量追求完整，但是从kafka造数据开始写，过于累赘因此省略。正文开始。

单表

kafka原始数据

{"id":1,"name":"nick","age":7,"address":"shanghai"}

原始数据形式

flinksql 连接

准备连接sql

    public static String kafkaTablePerson = "CREATE TABLE person (\n" +
            " id INT,\n" +
            " name STRING,\n" +
            " age INT,\n" +
            " address STRING\n" +
            ") WITH (\n" +
            " 'connector' = 'kafka',\n" +
            " 'topic' = 'flink',\n" +
            " 'properties.bootstrap.servers' = '172.8.10.xxx:6667',\n" +
            " 'format' = 'json',\n" +
            " 'scan.startup.mode' = 'earliest-offset'\n" +
            ")\n";

以下的连接器元数据可以在表定义中通过元数据列的形式获取。

R/W 列定义了一个元数据是可读的（R）还是可写的（W）。 只读列必须声明为 VIRTUAL 以在 INSERT INTO 操作中排除它们。

以下是kafka connector的参数，只写了常用的，文末有所有参数

连接器参数

flinksql查询

    public static String selectALLPerson = "select id,name,age,address from person ";

主体方法

  		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        tEnv.executeSql(CreateTableSQL.kafkaTableInfo);
    	tEnv.executeSql(SelectSQL.selectALLPerson).print();

简单的几行代码就可以搞定了，和写sql 没什么区别，十分便捷。

sql筛选语句

把sql稍微复杂一点点

public static String selectPerCountNamePrint = " select name, COUNT(*) as num FROM person GROUP BY name ";

依旧可以，不过有个坑，别名不能和方法名相同，这个在其他地方似乎没有这个限制的，我试了count(*) as count 就会出现错误。

数据落盘

仅仅只是select 是不会满足使用的，需要落盘。
我这里使用的是sink doris中

    public static final String SinkDorisPerCountName = "CREATE TABLE  percountname_sink (" +
            "name STRING, \n" +
            "num BIGINT  \n" +
            ")"+
            "WITH ( \n" +
            " 'connector' = 'doris' , \n" +
            " 'fenodes' = '172.8.10.xxx:8030' , \n" +
            " 'table.identifier' = 'test_db.PerCountName' ,\n " +
            " 'username' = 'username', \n " +
            " 'password' = 'password' ,\n " +
            "  'sink.label-prefix' ='" + label + "',\n" +
            "  'sink.properties.format' = 'json',\n" +
            "  'sink.properties.read_json_by_line' = 'true'\n" +
            ")";

需要注意的是label 必须是唯一的，为了避免多次测试时遇到问题，这里写了个随机的label，确保每次都不一样。

案例中的WITH 参数基本满足使用，更多细节推荐去官网看看。

为了方便我把doris建表贴出来

CREATE TABLE IF NOT EXISTS PerCountName
(
    `name` VARCHAR(50) NOT NULL COMMENT "",
    `num` INT NOT NULL COMMENT "出现的次数"
)
UNIQUE KEY(`name`)
DISTRIBUTED BY HASH(`name`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

insertSQL语句

public static String selectPerCountNamePrint = " select name, COUNT(*) as num FROM person GROUP BY name ";

执行

现在只需要把上文的主体方法稍微改一下即可

 		tEnv.executeSql(CreateTableSQL.kafkaTablePerson);
        tEnv.executeSql(SinkDorisTableSQL.SinkDorisPerCountName);
        tEnv.executeSql(SelectSQL.selectPerCountNameInsert);

执行的时候会发现doris里并没有数据，这是因为
为了保证Flink的Exactly Once语义，Flink Doris Connector 默认开启两阶段提交，Doris在1.1版本后默认开启两阶段提交。1.0可通过修改BE参数开启。
因此需要开启checkpoint 或者修改doris BE参数

 env.enableCheckpointing(10000);
 env.setParallelism(1);

加上之后即可

还去试了一下on yarn 是没有问题的，不过会遇到报错
可以看这篇文章
https://blog.csdn.net/weixin_45399602/article/details/127526911

好了，现在单表查询和落盘已经会了，想试一下多表查询有没有问题呢

多表join

新表结构这样

{"id":0,"name":"jack","score":141,"date":"2023-01-02"}

目标为：每个name的最大score。

简单一点 直接贴了

    public static String kafkaTableInfo = "CREATE TABLE info (\n" +
            " id INT,\n" +
            " name STRING,\n" +
            " score INT,\n" +
            " `date` DATE\n" +
            ") WITH (\n" +
            " 'connector' = 'kafka',\n" +
            " 'topic' = 'flinkinfo',\n" +
            " 'properties.bootstrap.servers' = '172.8.10.xxx:6667',\n" +
            " 'format' = 'json',\n" +
            " 'scan.startup.mode' = 'earliest-offset'\n" +
            ")\n";

 public static String selectPerMaxScore = " SELECT info.name, MAX(info.score) AS max_score FROM info JOIN person ON person.name = info.name GROUP BY info.name";

join查询

tEnv.executeSql(CreateTableSQL.kafkaTableInfo);
tEnv.executeSql(CreateTableSQL.kafkaTablePerson);
tEnv.executeSql(SelectSQL.selectPerMaxScore).print();

或者写到doris里 也是可以的，效果会比较明显。

总结

flinksql 消费kafka整体来说是十分简单好用的，使用时门口较低，实时性又高，可以经常使用。

OK，以上为案例所以内容。

以下是补充内容。

整个项目所需要的所有依赖

可以跳过文章来源地址https://www.toymoban.com/news/detail-630004.html

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <scala.version>2.12</scala.version>
    <java.version>1.8</java.version>
    <flink.version>1.14.4</flink.version>
    <fastjson.version>1.2.62</fastjson.version>
    <hadoop.version>2.8.3</hadoop.version>
    <scope.mode>compile</scope.mode>
    <slf4j.version>1.7.30</slf4j.version>

  </properties>

  <dependencies>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-table-api-scala-bridge_${scala.version}</artifactId>
          <version>${flink.version}</version>
          <scope>provided</scope>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-table-planner_${scala.version}</artifactId>
          <version>${flink.version}</version>
          <!--            本地测试时注释 scope -->
<!--          <scope>provided</scope>-->
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-streaming-scala_${scala.version}</artifactId>
          <version>${flink.version}</version>
          <!--            本地测试时注释 scope -->
<!--          <scope>provided</scope>-->
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-clients_${scala.version}</artifactId>
          <version>${flink.version}</version>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-connector-jdbc_${scala.version}</artifactId>
          <version>${flink.version}</version>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-connector-kafka_${scala.version}</artifactId>
          <version>${flink.version}</version>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-json</artifactId>
          <version>${flink.version}</version>
      </dependency>
      <dependency>
          <groupId>com.alibaba</groupId>
          <artifactId>fastjson</artifactId>
          <version>${fastjson.version}</version>
      </dependency>
      <!-- Add log dependencies when debugging locally -->
      <dependency>
          <groupId>org.slf4j</groupId>
          <artifactId>slf4j-api</artifactId>
          <version>${slf4j.version}</version>
      </dependency>
      <dependency>
          <groupId>org.slf4j</groupId>
          <artifactId>slf4j-log4j12</artifactId>
          <version>${slf4j.version}</version>
      </dependency>
      <!-- flink-doris-connector -->
      <dependency>
          <groupId>org.apache.doris</groupId>
          <artifactId>flink-doris-connector-1.14_2.12</artifactId>
          <version>1.1.0</version>
      </dependency>
      <dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <version>8.0.12</version>
      </dependency>
      <dependency>
          <groupId>com.ververica</groupId>
          <artifactId>flink-connector-mysql-cdc</artifactId>
          <version>2.2.0</version>
          <exclusions>
              <exclusion>
                  <artifactId>flink-shaded-guava</artifactId>
                  <groupId>org.apache.flink</groupId>
              </exclusion>
          </exclusions>
      </dependency>
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-runtime-web_${scala.version}</artifactId>
          <version>${flink.version}</version>
      </dependency>

      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
          <version>${flink.version}</version>
      </dependency>
      <!--保存检查点到hdfs上-->
      <dependency>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-client</artifactId>
          <version>${hadoop.version}</version>
      </dependency>

  </dependencies>

kafka connector参数

到了这里，关于FlinkSQL kafka完整案例 可直接复制使用的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！