注意！下述示例请使用我们修改编译好的包。

4.1 Kafka进Kafka出的简单ETL

4.1.1 需求

对test_csv主题中的数据进行过滤，仅保留年龄在18岁以上的记录。

4.1.2 需求实现

1）首先，创建为kafka创建test_csv主题。

kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic test_csv --partitions 1 --replication-factor 1

2）为kafka创建test_sink主题

kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic test_sink --partitions 1 --replication-factor 1

3）编辑应用配置

[atguigu@hadoop102 config]$ vim example03.conf

4）应用配置内容

env {
  # You can set flink configuration here
  execution.parallelism = 1
  #execution.checkpoint.interval = 10000
  #execution.checkpoint.data-uri = "hdfs://hadoop102:9092/checkpoint"
}
 
# 在source所属的块中配置数据源
source {
    KafkaTableStream {
        consumer.bootstrap.servers = "hadoop102:9092"
        consumer.group.id = "seatunnel-learn"
        topics = test_csv
        result_table_name = test
        format.type = csv
        schema = "[{\"field\":\"name\",\"type\":\"string\"},{\"field\":\"age\", \"type\": \"int\"}]"
        format.field-delimiter = ";"
        format.allow-comments = "true"
        format.ignore-parse-errors = "true"
    }
}
# 在transform的块中声明转换插件
transform {
 
  sql {
    sql = "select name,age from test  where age > '"${age}"'"
  }
}
# 在sink块中声明要输出到哪
sink {
   kafkaTable {
    topics = "test_sink"
    producer.bootstrap.servers = "hadoop102:9092"
        }
}

5）提交任务

bin/start-seatunnel-flink.sh --config config/example03.conf -i age=18

6）起一个kafka console producer 发送csv数据(分号分隔)

7）起一个kafka console consumer 消费数据

我们成功地实现了数据从kafka输入经过简单的ETL再向kafka输出。

4.2 Kafka 输出到Doris进行指标统计

4.2.1 需求

使用回话日志统计用户的总观看视频数，用户最常会话市场，用户最小会话时长，用户最后一次会话时间。

4.2.2 需求实现

1）在资料中有一个伪数据的生成脚本，将它拷贝到服务器的任意位置

2）执行以下命令安装python脚本需要的两个依赖库

pip3 install Faker
pip3 install kafka-python

3）使用mysql客户端连接doris

[atguigu@hadoop102 fake_data]$ mysql -h hadoop102 -P 9030 -uatguigu  -p123321

4）手动创建test_db数据库。

create database test_db;

5）使用下述sql语句建表

CREATE TABLE `example_user_video` (
  `user_id` largeint(40) NOT NULL COMMENT "用户id",
  `city` varchar(20) NOT NULL COMMENT "用户所在城市",
  `age` smallint(6) NULL COMMENT "用户年龄",
  `video_sum` bigint(20) SUM NULL DEFAULT "0" COMMENT "总观看视频数",
  `max_duration_time` int(11) MAX NULL DEFAULT "0" COMMENT "用户最长会话时长",
  `min_duration_time` int(11) MIN NULL DEFAULT "999999999" COMMENT "用户最小会话时长",
  `last_session_date` datetime REPLACE NULL DEFAULT "1970-01-01 00:00:00" COMMENT "用户最后一次会话时间"
) ENGINE=OLAP
AGGREGATE KEY(`user_id`, `city`, `age`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
;                    
6）在config目录下， 编写如下的配置文件。
env {
 execution.parallelism = 1
}
 
source {
    KafkaTableStream {
     consumer.bootstrap.servers = "hadoop102:9092"
     consumer.group.id = "seatunnel5"
     topics = test
     result_table_name = test
     format.type = json
     schema = "{\"session_id\":\"string\",\"video_count\":\"int\",\"duration_time\":\"long\",\"user_id\":\"string\",
     \"user_age\":\"int\",\"city\":\"string\",\"session_start_time\":\"datetime\",\"session_end_time\":\"datetime\"}"
     format.ignore-parse-errors = "true"
 }
}
 
transform{
 sql {
  sql = "select user_id,city,user_age as age,video_count as video_sum,duration_time as max_duration_time,duration_time 
  as min_duration_time,session_end_time as last_session_date from test"
  result_table_name = test2
 }
}
 
sink{
 DorisSink {
  source_table_name = test2
     fenodes = "hadoop102:8030"
     database = test_db
     table = example_user_video
     user = atguigu
     password = 123321
     batch_size = 50
     doris.column_separator="\t"
     doris.columns="user_id,city,age,video_sum,max_duration_time,min_duration_time,last_session_date"
 }
}

7）使用python脚本向kafka中生成伪数据

[atguigu@hadoop102 fake_data]$ python3 fake_video.py --bootstrap-server hadoop102:9092 --topic test_video

8）查看doris中的结果。

Select * from `example_user_video`;