运行环境配置

集群配置

  1. execution.checkpointing.unaligned=true
  2. execution.checkpointing.interval=10 min
  4. containerized.taskmanager.env.PYTHONPATH=/home/appweb/shuidi/software/python

Python配置

  1. import random
  3. @udf(result_type=DataTypes.STRING(), func_type="pandas")
  4. def generate_url(column):
  5.     urls = [
  6.         'http://iceye.wacai.info/stream-schema/info',
  7.         'http://druid.router.wacai.info/unified-console.html#query',
  8.         'http://merlin-wdp.wacai.info/flink/job/list',
  9.         'http://cdhmanager.stanlee.wacai.info/cmf/login',
  10.         'http://middleware.wacai.info/hermes-web/dashboard',
  11.         'http://streaming-yarn.stanlee.wacai.info/cluster/apps/RUNNING',
  12.         'http://team.caimi-inc.com/',
  13.         'https://loan.wacai.info/wups/roleApply',
  14.         'http://mario.wacai.info/home#',
  15.         'http://biuc.wacai.info/',
  16.         'https://logs.wacai.info/app/kibana#/home?_g=()',
  17.         'http://awm.wacai.info/awm/notebook',
  18.         'http://grafana.wacai.info/?orgId=1',
  19.         'http://wafe.qsh.wacai.info/configMgr',
  20.         'http://sensordata.wacai.info/login/index.html',
  21.         'http://zeppelin.wacai.info/#/',
  22.         'http://prometheus-qsh.wacai.info/graph',
  23.         'http://bitask.caimi-inc.com/index.php?m=script&a=index#',
  24.         'http://platform.wacai.info/prophet-node/job',
  25.         'http://atlas.wacai.info/index.html#!/glossary'
  26.     ]
  27.     return column.map(lambda item: random.choice(urls))
  29. st_env.drop_temporary_function('generate_url')
  30. st_env.create_temporary_function('generate_url', generate_url)

SQL配置

SQL运行配置

  1. -- SQL运行配置
  2. SET table.dynamic-table-options.enabled=true; -- 开启 SQL hints功能
  3. SET table.exec.mini-batch.enabled=true; -- 开启mini-batch
  4. SET table.exec.mini-batch.allow-latency=5 s;
  5. SET table.exec.mini-batch.size=5000;
  6. SET table.exec.state.ttl=48 h; -- 设置State保存时长
  7. SET python.executable=/data/program/miniconda3/envs/pyflink/bin/python; -- 设计python运行程序路径

创建Catalog

  1. CREATE CATALOG iceberg WITH (
  2.   'type'='iceberg',
  3.   'catalog-type'='hive',
  4.   'uri'='thrift://172.16.48.191:9083',
  5.   'clients'='5',
  6.   'property-version'='1',
  7.   'hive-conf-dir'='/data/program/flink-114/conf/iceberg'
  8. );
  10. -- flink jdbc url参数配置
  11. -- jdbc:mysql://192.168.5.12:3307?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&rewriteBatchedStatements=true&useServerPrepStmts=true&cachePrepStmts=true&allowMultiQueries=true&useConfigs=maxPerformance
  12. CREATE CATALOG jdbc WITH(
  13.   'type' = 'jdbc',
  14.   'default-database' = 'data_dict',
  15.   'username' = 'qa_conn',
  16.   'password' = 'qa_conn',
  17.   'base-url' = 'jdbc:mysql://192.168.5.12:3307'
  18. );
  20. CREATE CATALOG mysql_cdc WITH(
  21.   'type' = 'mysql-cdc',
  22.   'default-database' = 'wac_edu',
  23.   'hostname' = '192.168.5.14',
  24.   'port' = '3342',
  25.   'username' = 'qa_conn',
  26.   'password' = 'qa_conn'
  27. );

创建基本的表

  1. -- =================================
  2. -- 定义Source，View
  3. -- =================================
  4. -- 生成数据
  5. CREATE TABLE clicks_datagen (
  6.     `user` VARCHAR COMMENT '用户名称',
  7.     `cTime` AS LOCALTIMESTAMP COMMENT '点击时间',
  8.     `url` VARCHAR COMMENT '点击链接',
  9.     `procTime` AS PROCTIME() COMMENT '处理时间',
  10.     WATERMARK FOR `cTime` AS `cTime` - INTERVAL '5' SECOND
  11. )
  12. COMMENT '用户点击表'
  13. WITH (
  14.     'connector' = 'datagen',
  15.     'rows-per-second' = '100',
  16.     'fields.user.kind' = 'random',
  17.     'fields.user.length' = '5',
  18.     'fields.url.kind' = 'random',
  19.     'fields.url.length' = '5'
  20. );
  22. -- 转化生成的数据
  23. CREATE VIEW clicks_datagen_view AS
  24. SELECT
  25.     `user`,
  26.     `cTime`,
  27.     generate_url(`url`) as `url`,
  28.     `procTime`
  29. FROM clicks_datagen;
  31. -- blackhole 写入这里的数据都会ignore
  32. CREATE TABLE clicks_blackhole (
  33.     `user` VARCHAR COMMENT '用户名称',
  34.     `cTime` TIMESTAMP COMMENT '点击时间',
  35.     `url` VARCHAR COMMENT '点击链接'
  36. )
  37. COMMENT '用户点击表'
  38. WITH (
  39.     'connector' = 'blackhole'
  40. );

datagen —> hermes —> blackhole

  1. CREATE TABLE clicks_hermes (
  2.     `user` VARCHAR COMMENT '用户名称',
  3.     `cTime` TIMESTAMP COMMENT '点击时间',
  4.     `url` VARCHAR COMMENT '点击链接'
  5. )
  6. COMMENT '用户点击表'
  7. WITH (
  8.     'connector' = 'hermes',
  9.     'properties.hermes.center' = 'http://hermes-center.middleware.wse.test.wacai.info',
  10.     'properties.hermes.cluster.id' = 'test',
  11.     'properties.group.id' = 'flink-test',
  12.     'scan.startup.mode' = 'latest-offset',
  13.     'topic' = 'merlin.flink.clicks',
  14.     'format' = 'json'
  15. );
  17. -- datagen --> hermes --> blackhole
  18. INSERT INTO clicks_hermes
  19. SELECT
  20.     `user`,
  21.     `cTime`,
  22.     `url`
  23. FROM clicks_datagen_view;
  25. INSERT INTO clicks_blackhole
  26. SELECT * FROM clicks_hermes;

datagen —> kafka sink —> kafka raw source —> hdfs

  1. -- kafka topic
  2. CREATE TABLE clicks_kafka (
  3.     `user` VARCHAR COMMENT '用户名称',
  4.     `cTime` TIMESTAMP COMMENT '点击时间',
  5.     `url` VARCHAR COMMENT '点击链接'
  6. )
  7. COMMENT '用户点击表'
  8. WITH (
  9.     'connector' = 'kafka',
  10.     'properties.bootstrap.servers' = '172.16.48.194:9092',
  11.     'properties.transactional.id' = 'flink1.11_test-transaction',
  12.     'properties.acks' = 'all',
  13.     'properties.isolation.level' = 'read_committed',
  14.     'properties.group.id' = 'flink-test',
  15.     'scan.startup.mode' = 'latest-offset',
  16.     'topic' = 'merlin.flink.clicks',
  17.     'format' = 'json'
  18. );
  20. -- kafka topic，format格式是raw,不解析内容
  21. CREATE TABLE clicks_kafka_source_raw (
  22.     `content` BYTES,
  23.     `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
  24.     WATERMARK FOR `ts` AS `ts` - INTERVAL '5' SECOND
  25. )
  26. COMMENT '用户点击表'
  27. WITH (
  28.     'connector' = 'kafka',
  29.     'properties.bootstrap.servers' = '172.16.48.194:9092',
  30.     'properties.transactional.id' = 'flink1.11_test-transaction',
  31.     'properties.acks' = 'all',
  32.     'properties.isolation.level' = 'read_committed',
  33.     'properties.group.id' = 'flink-test',
  34.     'scan.startup.mode' = 'latest-offset',
  35.     'topic' = 'merlin.flink.clicks',
  36.     'format' = 'raw'
  37. );
  39. -- hdfs sink
  40. CREATE TABLE clicks_hdfs (
  41.     `content` BYTES,
  42.     `dt` VARCHAR,
  43.     `hour` VARCHAR
  44. )
  45. COMMENT '用户点击表'
  46. PARTITIONED BY (`dt`, `hour`) WITH (
  47.     'connector' = 'filesystem',
  48.     'path' = 'hdfs://172.16.48.23:8020/shuidi/flink/tables/clicks',
  49.     'sink.partition-commit.trigger' = 'partition-time',
  50.     'sink.partition-commit.delay' = '1 h',
  51.     'sink.rolling-policy.rollover-interval' = '60 min',
  52.     'partition.time-extractor.timestamp-pattern' = '$dt $hour:00:00',
  53.     'sink.partition-commit.policy.kind' = 'success-file',
  54.     'sink.partition-commit.success-file.name' = '_success',
  55.     'format' = 'raw'
  56. );
  58. -- datagen --> kafka sink --> kafka raw source --> hdfs
  59. INSERT INTO clicks_kafka
  60. SELECT
  61.     `user`,
  62.     `cTime`,
  63.     `url`
  64. FROM clicks_datagen_view;
  66. INSERT INTO clicks_hdfs
  67. SELECT
  68.     `content`,
  69.     CAST(CAST(`ts` AS DATE) AS VARCHAR),
  70.     CAST(HOUR(`ts`) AS VARCHAR)
  71. FROM clicks_kafka_source_raw;

datagen —> pulsar

  1. -- pulsar sink
  2. CREATE TABLE clicks_pulsar (
  3.     `user` VARCHAR COMMENT '用户名称',
  4.     `cTime` VARCHAR COMMENT '点击时间',
  5.     `url` VARCHAR COMMENT '点击链接'
  6. )
  7. COMMENT '用户点击表'
  8. WITH (
  9.     'connector' = 'pulsar',
  10.     'topic' = 'persistent://merlin/flink-sql/clicks',
  11.     'service-url' = 'pulsar://172.16.48.194:6650',
  12.     'admin-url' = 'http://172.16.48.194:8080',
  13.     'format' = 'json'
  14. );
  16. -- datagen --> pulsar
  17. INSERT INTO clicks_pulsar
  18. SELECT
  19.     `user`,
  20.     CAST(`cTime` AS VARCHAR),
  21.     `url`
  22. FROM clicks_datagen_view;

datagen —> elasticsearch

  1. -- elasticsearch 6
  2. -- =================================
  3. -- elasticsearch sink
  4. CREATE TABLE clicks_es6(
  5.     `user` VARCHAR COMMENT '用户名称',
  6.     `cTime` TIMESTAMP COMMENT '点击时间',
  7.     `url` VARCHAR COMMENT '点击链接'
  8. ) WITH (
  9.     'connector' = 'elasticsearch-6',
  10.     'hosts' = 'http://192.168.5.16:9200;http://192.168.5.17:9200',
  11.     'index' = 'merlin_flink_clicks-{cTime|yyyy-MM-dd}'
  12. );
  14. -- datagen --> kafka --> es6
  15. INSERT INTO clicks_es6
  16. SELECT
  17.     `user`,
  18.     `cTime`,
  19.     `url`
  20. FROM clicks_kafka;
  23. -- elasticsearch 7
  24. -- =================================
  25. -- elasticsearch sink
  26. CREATE TABLE clicks_es7(
  27.     `user` VARCHAR COMMENT '用户名称',
  28.     `cTime` TIMESTAMP COMMENT '点击时间',
  29.     `url` VARCHAR COMMENT '点击链接'
  30. ) WITH (
  31.     'connector' = 'elasticsearch-7',
  32.     'hosts' = 'http://172.16.48.194:9200',
  33.     'index' = 'merlin_flink_clicks-{cTime|yyyy-MM-dd}'
  34. );
  36. -- datagen --> kafka --> es7
  37. INSERT INTO clicks_es7
  38. SELECT
  39.     `user`,
  40.     `cTime`,
  41.     `url`
  42. FROM clicks_kafka;

datagen —> kafka upsert —> blackhole

  1. CREATE TABLE clicks_upsert_kafka (
  2.     `user` VARCHAR COMMENT '用户名称',
  3.     `cTime` TIMESTAMP COMMENT '点击时间',
  4.     `url` VARCHAR COMMENT '点击链接',
  5.     PRIMARY KEY (`user`) NOT ENFORCED
  6. )
  7. COMMENT '用户点击表'
  8. WITH (
  9.     'connector' = 'upsert-kafka',
  10.     'properties.bootstrap.servers' = '172.16.48.194:9092',
  11.     'properties.group.id' = 'flink-test',
  12.     'topic' = 'merlin.flink.clicks.upsert',
  13.     'key.format' = 'json',
  14.     'value.format' = 'json',
  15.     'value.fields-include' = 'EXCEPT_KEY'
  16. );
  18. -- datagen --> kafka upsert --> blackhole
  19. INSERT INTO clicks_upsert_kafka
  20. SELECT
  21.     `user`,
  22.     `cTime`,
  23.     `url`
  24. FROM clicks_datagen_view;
  26. INSERT INTO clicks_blackhole
  27. SELECT * FROM clicks_upsert_kafka;

更新数据实例

按天统计更新数据

SQL

  1. CREATE TABLE teacher_student_relation_source(
  2.     `id` BIGINT COMMENT '主键id',
  3.     `teacher_id` BIGINT COMMENT '讲师id',
  4.     `student_id` BIGINT COMMENT '学生id',
  5.     `status` INT COMMENT '状态，1表示绑定，0表示解绑',
  6.     `talk_times` INT COMMENT '交流次数',
  7.     `relation_level` INT COMMENT '关系等级，值越高等级越强',
  8.     `created_time` TIMESTAMP COMMENT '创建时间',
  9.     `updated_time` TIMESTAMP COMMENT '更新时间',
  10.     WATERMARK FOR `updated_time` AS `updated_time`
  11. ) WITH (
  12.     'connector' = 'kafka',
  13.     'topic' = 'teacher.student.relation',
  14.     'properties.bootstrap.servers' = '172.16.48.194:9092',
  15.     'properties.group.id' = 'flink-test',
  16.     'scan.startup.mode' = 'earliest-offset',
  17.     'format' = 'debezium-json'
  18. );
  20. CREATE TABLE print_sink(
  21.     `dt` DATE COMMENT '日期',
  22.     `teacher_id` BIGINT COMMENT '讲师id',
  23.     `cnt` BIGINT COMMENT '个数'
  24. ) WITH (
  25.     'connector' = 'print'
  26. );
  28. INSERT INTO print_sink
  29. SELECT
  30.     CAST(`updated_time` AS DATE) AS `dt`,
  31.     `teacher_id`,
  32.     count(*) AS cnt
  33. FROM teacher_student_relation_source
  34. WHERE `status` = 1
  35. GROUP BY CAST(`updated_time` AS DATE), `teacher_id`;

输入

  1. {"after":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 11:21:10"},"op":"c"}
  2. {"after":{"id":2,"teacher_id":1,"student_id":2,"status":1,"talk_times":2,"relation_level":2,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 11:21:10"},"op":"c"}
  3. {"after":{"id":3,"teacher_id":1,"student_id":3,"status":1,"talk_times":3,"relation_level":3,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 11:21:10"},"op":"c"}
  4. {"before":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 11:21:10"},"after":{"id":1,"teacher_id":1,"student_id":1,"status":0,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 13:21:10"},"op":"u"}
  5. {"before":{"id":1,"teacher_id":1,"student_id":1,"status":0,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 13:21:10"},"after":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 14:21:10"},"op":"u"}
  6. {"before":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":1,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 14:21:10"},"after":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":2,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 14:22:10"},"op":"u"}
  7. {"before":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":2,"relation_level":1,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 14:22:10"},"after":{"id":1,"teacher_id":1,"student_id":1,"status":1,"talk_times":100,"relation_level":2,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 14:40:10"},"op":"u"}
  8. {"before":{"id":2,"teacher_id":1,"student_id":2,"status":1,"talk_times":2,"relation_level":2,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 15:21:10"},"op":"d"}
  9. {"before":{"id":3,"teacher_id":1,"student_id":3,"status":1,"talk_times":3,"relation_level":3,"created_time":"2021-09-30 11:21:10","updated_time":"2021-09-30 11:21:10"},"after":{"id":3,"teacher_id":1,"student_id":3,"status":1,"talk_times":3,"relation_level":3,"created_time":"2021-09-30 11:21:10","updated_time":"2021-10-01 11:21:10"},"op":"u"}

输出

  1. +I(2021-09-30,1,1)
  3. -U(2021-09-30,1,1)
  4. +U(2021-09-30,1,2)
  6. -U(2021-09-30,1,2)
  7. +U(2021-09-30,1,3)
  9. -U(2021-09-30,1,3)
  10. +U(2021-09-30,1,2)
  12. -U(2021-09-30,1,2)
  13. +U(2021-09-30,1,3)
  15. -U(2021-09-30,1,3)
  16. +U(2021-09-30,1,2)
  17. -U(2021-09-30,1,2)
  18. +U(2021-09-30,1,3)
  20. -U(2021-09-30,1,3)
  21. +U(2021-09-30,1,2)
  22. -U(2021-09-30,1,2)
  23. +U(2021-09-30,1,3)
  25. -U(2021-09-30,1,3)
  26. +U(2021-09-30,1,2)
  28. -U(2021-09-30,1,2)
  29. +U(2021-09-30,1,1)
  30. +I(2021-10-01,1,1)