0
0

使用Flink MySQL cdc分别sink到ES、Kafka、Hudi

环境说明

版本问题

flink cdc与flink对应版本

https://ververica.github.io/flink-cdc-connectors/master/content/about.html#supported-flink-versions

hudi与flink版本

https://github.com/apache/hudi/issues/3689

编译所需jar包

编译flink-cdc-connectors

  • 进入flink-cdc-connectors目录，修改flink-cdc-connectors/pom.xml文件中flink的版本号为1.13.1

<flink.version>1.13.1</flink.version>

- maven编译

mvn clean package -DskipTests

- 编译完成，将以下jar包拷贝到flink-1.13.1/lib下

flink-cdc-connectors/flink-sql-connector-mysql-cdc/target/flink-sql-connector-mysql-cdc-2.1-SNAPSHOT.jar

flink-cdc-connectors/flink-sql-connector-postgres-cdc/target/flink-sql-connector-postgres-cdc-2.1-SNAPSHOT.jar

flink-cdc-connectors/flink-format-changelog-json/target/flink-format-changelog-json-2.1-SNAPSHOT.jar

编译hudi

  • 进入hudi目录，修改hudi/pom.xml，修改对应组件的版本，由于flink使用的是scala-2.11版本，spark3.x版本以上默认使用scala-2.12预编译，为了节省时间，我们在此使用spark2.4.8以scala-2.11预编译的版本，对应的hadoop版本为2.7

<flink.version>1.13.1</flink.version>
<hadoop.version>2.7.3</hadoop.version>
<spark2.version>2.4.8</spark2.version>

- maven编译

mvn clean package -DskipTests

- 编译完成，将以下jar包拷贝到flink-1.13.1/lib下

hudi/packaging/hudi-flink-bundle/target/hudi-flink-bundle_2.11-0.10.0-SNAPSHOT.jar

- 将以下jar包拷贝到spark-2.4.8-bin-hadoop2.7/jars下

hudi/packaging/hudi-spark-bundle/target/hudi-spark-bundle_2.11-0.10.0-SNAPSHOT.jar

- 将以下jar包拷贝到hadoop-2.7.3/share/hadoop/hdfs/下

hudi/packaging/hudi-hadoop-mr-bundle/target/hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar

- 另外为了把数据写入ES和kafka中，需要下载以下jar包，并拷贝到flink-1.13.1/lib下

flink-sql-connector-elasticsearch7_2.11-1.13.1.jar

flink-sql-connector-kafka_2.11-1.13.1.jar

- flink sink到hudi还需要以下jar包

flink-shaded-hadoop-2-uber-2.7.5-10.0.jar

- 所有依赖放置完成后的flink-1.13.1/lib如下

Docker中创建所需的组件

  • 创建一个目录，创建一个新文件docker-compose.yml，内容如下,包含postgres mysql es kibana kafka zookeeper组件：
  2. version: '2.1'
  3. services:
  4.   postgres:
  5.     image: debezium/example-postgres:1.1
  6.     ports:
  7.       - "5432:5432"
  8.     environment:
  9.       - POSTGRES_PASSWORD=1234
  10.       - POSTGRES_DB=postgres
  11.       - POSTGRES_USER=postgres
  12.       - POSTGRES_PASSWORD=postgres
  13.   mysql:
  14.     image: debezium/example-mysql:1.1
  15.     ports:
  16.       - "3306:3306"
  17.     environment:
  18.       - MYSQL_ROOT_PASSWORD=123456
  19.       - MYSQL_USER=mysqluser
  20.       - MYSQL_PASSWORD=mysqlpw
  21.   elasticsearch:
  22.     image: elastic/elasticsearch:7.6.0
  23.     environment:
  24.       - cluster.name=docker-cluster
  25.       - bootstrap.memory_lock=true
  26.       - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
  27.       - discovery.type=single-node
  28.     ports:
  29.       - "9200:9200"
  30.       - "9300:9300"
  31.     ulimits:
  32.       memlock:
  33.         soft: -1
  34.         hard: -1
  35.       nofile:
  36.         soft: 65536
  37.         hard: 65536
  38.   kibana:
  39.     image: elastic/kibana:7.6.0
  40.     ports:
  41.       - "5601:5601"
  42.   zookeeper:
  43.     image: wurstmeister/zookeeper:3.4.6
  44.     ports:
  45.       - "2181:2181"
  46.   kafka:
  47.     image: wurstmeister/kafka:2.12-2.2.1
  48.     ports:
  49.       - "9092:9092"
  50.       - "9094:9094"
  51.     depends_on:
  52.       - zookeeper
  53.     environment:
  54.       - KAFKA_ADVERTISED_LISTENERS=INSIDE://:9094,OUTSIDE://localhost:9092
  55.       - KAFKA_LISTENERS=INSIDE://:9094,OUTSIDE://:9092
  56.       - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
  57.       - KAFKA_INTER_BROKER_LISTENER_NAME=INSIDE
  58.       - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
  59.       - KAFKA_CREATE_TOPICS="user_behavior:1:1"
  60.     volumes:
  61.       - /var/run/docker.sock:/var/run/docker.sock

- 进入目录，启动并初始化所有容器

docker-compose build
docker-compose up -d

- 打开docker 应用，可以看到组件已经全部启动

初始化数据

- 通过docker命令进入mysql容器内部

docker-compose exec mysql mysql -uroot -p123456

- 初始化MySQL数据

  2. -- MySQL
  3. CREATE DATABASE mydb;
  4. USE mydb;
  5. CREATE TABLE products (
  6.   id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  7.   name VARCHAR(255) NOT NULL,
  8.   description VARCHAR(512)
  9. );
  10. ALTER TABLE products AUTO_INCREMENT = 101;
  12. INSERT INTO products
  13. VALUES (default,"scooter","Small 2-wheel scooter"),
  14.        (default,"car battery","12V car battery"),
  15.        (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
  16.        (default,"hammer","12oz carpenter's hammer"),
  17.        (default,"hammer","14oz carpenter's hammer"),
  18.        (default,"hammer","16oz carpenter's hammer"),
  19.        (default,"rocks","box of assorted rocks"),
  20.        (default,"jacket","water resistent black wind breaker"),
  21.        (default,"spare tire","24 inch spare tire");
  23. CREATE TABLE orders (
  24.   order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  25.   order_date DATETIME NOT NULL,
  26.   customer_name VARCHAR(255) NOT NULL,
  27.   price DECIMAL(10, 5) NOT NULL,
  28.   product_id INTEGER NOT NULL,
  29.   order_status BOOLEAN NOT NULL -- 是否下单
  30. ) AUTO_INCREMENT = 10001;
  32. INSERT INTO orders
  33. VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
  34.        (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
  35.        (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

- 通过docker命令进入postgres容器内部

docker-compose exec postgres psql -h localhost -U postgres

- 初始化postgres数据

  1. -- PG
  2. CREATE TABLE shipments (
  3.   shipment_id SERIAL NOT NULL PRIMARY KEY,
  4.   order_id SERIAL NOT NULL,
  5.   origin VARCHAR(255) NOT NULL,
  6.   destination VARCHAR(255) NOT NULL,
  7.   is_arrived BOOLEAN NOT NULL
  8. );
  9. ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001;
  10. ALTER TABLE public.shipments REPLICA IDENTITY FULL;
  12. INSERT INTO shipments
  13. VALUES (default,10001,'Beijing','Shanghai',false),
  14.        (default,10002,'Hangzhou','Shanghai',false),
  15.        (default,10003,'Shanghai','Hangzhou',false);

修改环境变量、hadoop配置文件并启动hadoop

- 修改环境变量vim ~/.zshrc 或者 vim /etc/profile

  2. export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8/Contents/Home
  3. export PATH=$PATH:$JAVA_HOME/bin
  5. export SCALA_HOME=/xxx/xxx/xxx/xxx/xxx/xxx/scala
  6. export PATH=$PATH:$SCALA_HOME/bin
  8. export SPARK_HOME=/xxx/xxx/xxx/xxx/xxx/xxx/spark
  9. export PATH=$PATH:$SPARK_HOME/bin
  11. export MAVEN_HOME=/xxx/xxx/xxx/xxx/xxx/xxx/apache-maven
  12. export PATH=$PATH:$MAVEN_HOME/bin
  14. export HADOOP_HOME=/xxx/xxx/xxx/xxx/xxx/xxx/hadoop
  15. export PATH=$PATH:$HADOOP_HOME/bin
  17. export FLINK_HOME=/xxx/xxx/xxx/xxx/xxx/xxx/flink
  18. export PATH=$PATH:$FLINK_HOME/bin
  20. export PATH=/xxx/xxx/xxx/xxx/xxx/xxx/protobuf/bin:$PATH

修改完成执行命令 source ~/.zshrc 或者source /etc/profile让环境变量生效

- 修改hadoop-2.7.3/etc/hadoop/core-site.xml，如果目录不存在，给创建上

  1. <configuration>
  2. <property>
  3.      <name>hadoop.tmp.dir</name> 
  4.      <value>file:/xxx/xxx/xxx/xxx/xxx/hadoop-2.7.3/tmp</value>
  5.   </property>
  6.   <property>
  7.      <name>fs.defaultFS</name>
  8.      <value>hdfs://localhost:8020</value>
  9.   </property>
  10. </configuration>
  • 修改hadoop-2.7.3/etc/hadoop/hdfs-site.xml，如果目录不存在，给创建上
  2. <configuration>
  3. <property>
  4.          <name>dfs.replication</name>
  5.          <value>1</value>
  6.     </property>
  7.     <property> 
  8.          <name>dfs.namenode.name.dir</name>
  9.          <value>file:/xxx/xxx/xxx/xxx/xxx/hadoop-2.7.3/dfs/name</value>
  10.     </property>
  11.     <property>
  12.          <name>dfs.namenode.data.dir</name>
  13.          <value>file:/xxx/xxx/xxx/xxx/xxx/hadoop-2.7.3/dfs/data</value>
  14.     </property>
  15. </configuration>
  • 修改hadoop-2.7.3/etc/hadoop/yarn-site.xml
  2. <configuration>
  4. <!-- Site specific YARN configuration properties -->
  5. <property>
  6.         <name>yarn.nodemanager.aux-services</name>
  7.         <value>mapreduce_shuffle</value>
  8. </property>
  9. <property>
  10.         <name>yarn.nodemanager.env-whitelist</name>
  11.  <value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAPRED_HOME</value>
  12. </property>
  14. </configuration>
  • 修改hadoop-2.7.3/etc/hadoop/mapred-site.xml
  2. <configuration>
  3. <property> 
  4. <name>mapreduce.framework.name
  5. </name>
  6.  <value>yarn</value>
  7.  </property>
  8. </configuration>
  • 启动,hadoop目录下执行
  2. ./sbin/start-all.sh
  3. # 格式化namenode
  4. hdfs namenode -format

如果不能发现Java环境变量，建议直接修改hadoop-2.7.3/etc/hadoop/hadoop-env.sh

  2. # export JAVA_HOME=${JAVA_HOME}
  3. export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8/Contents/Home
  • 启动完成，通过jps命令查看组件是否已启动,NameNode、SecondaryNameNode、DataNode、ResourceManager、NodeManager均已启动，启动完成，可以登录浏览器：http://localhost:50070/查看hdfs状态

➜ hadoop-2.7.3 jps
63696 NodeManager
63969 Jps
62178 DataNode
62805 SecondaryNameNode
63510 ResourceManager
2392
61789 NameNode

修改flink配置文件、并启动

  • 修改flink-1.13.1/conf/flink-conf.yaml，增加slots数量，修改jobmanager的端口

taskmanager.numberOfTaskSlots: 1

taskmanager.numberOfTaskSlots: 6

  • 启动flink

./bin/start-cluster.sh

➜ flink-1.13.1 jps
2320 SecondaryNameNode
3137 ResourceManager
3489 NodeManager
85906 TaskManagerRunner
16275 NameNode
1813 DataNode
2392
85565 StandaloneSessionClusterEntrypoint
86431 Jps

  • 启动flink的sql client

./bin/sql-client.sh

创建flink cdc表,分别以mysql、pgsql作为数据源，将数据写入到es中和kafka中

分别创建mysql和pgsql的flink-sql表

  2. --Flink SQL
  3. -- 设置 checkpoint 间隔为 3 秒                     
  4. Flink SQL> SET execution.checkpointing.interval = 3s;
  5. Flink SQL> set dfs.client.block.write.replace-datanode-on-failure.enable = ture;
  6. Flink SQL> set dfs.client.block.write.replace-datanode-on-failure.policy = NEVER;
  7. Flink SQL> CREATE TABLE products (
  8.    id INT,
  9.    name STRING,
  10.    description STRING,
  11.    PRIMARY KEY (id) NOT ENFORCED
  12.  ) WITH (
  13.    'connector' = 'mysql-cdc',
  14.    'hostname' = 'localhost',
  15.    'port' = '3306',
  16.    'username' = 'root',
  17.    'password' = '123456',
  18.    'database-name' = 'mydb',
  19.    'table-name' = 'products'
  20.  );
  22. Flink SQL> CREATE TABLE orders (
  23.    order_id INT,
  24.    order_date TIMESTAMP(0),
  25.    customer_name STRING,
  26.    price DECIMAL(10, 5),
  27.    product_id INT,
  28.    order_status BOOLEAN,
  29.    PRIMARY KEY (order_id) NOT ENFORCED
  30.  ) WITH (
  31.    'connector' = 'mysql-cdc',
  32.    'hostname' = 'localhost',
  33.    'port' = '3306',
  34.    'username' = 'root',
  35.    'password' = '123456',
  36.    'database-name' = 'mydb',
  37.    'table-name' = 'orders'
  38.  );
  40. Flink SQL> CREATE TABLE shipments (
  41.    shipment_id INT,
  42.    order_id INT,
  43.    origin STRING,
  44.    destination STRING,
  45.    is_arrived BOOLEAN,
  46.    PRIMARY KEY (shipment_id) NOT ENFORCED
  47.  ) WITH (
  48.    'connector' = 'postgres-cdc',
  49.    'hostname' = 'localhost',
  50.    'port' = '5432',
  51.    'username' = 'postgres',
  52.    'password' = 'postgres',
  53.    'database-name' = 'postgres',
  54.    'schema-name' = 'public',
  55.    'table-name' = 'shipments'
  56.  );

sink到es

  2. Flink SQL> CREATE TABLE enriched_orders (
  3.    order_id INT,
  4.    order_date TIMESTAMP(0),
  5.    customer_name STRING,
  6.    price DECIMAL(10, 5),
  7.    product_id INT,
  8.    order_status BOOLEAN,
  9.    product_name STRING,
  10.    product_description STRING,
  11.    shipment_id INT,
  12.    origin STRING,
  13.    destination STRING,
  14.    is_arrived BOOLEAN,
  15.    PRIMARY KEY (order_id) NOT ENFORCED
  16.  ) WITH (
  17.      'connector' = 'elasticsearch-7',
  18.      'hosts' = 'http://localhost:9200',
  19.      'index' = 'enriched_orders'
  20.  );
  22. Flink SQL> INSERT INTO enriched_orders
  23.  SELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrived
  24.  FROM orders AS o
  25.  LEFT JOIN products AS p ON o.product_id = p.id
  26.  LEFT JOIN shipments AS s ON o.order_id = s.order_id;

  • 修改 mysql 和 postgres 里面的数据，观察 elasticsearch 里的结果
  2. -- 修改MySQL
  3. INSERT INTO orders
  4. VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);
  6. -- 修改PG
  7. INSERT INTO shipments
  8. VALUES (default,10004,'Shanghai','Beijing',false);
  10. -- 修改MySQL
  11. UPDATE orders SET order_status = true WHERE order_id = 10004;
  13. -- 修改PG
  14. UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;
  16. -- 修改MySQL
  17. DELETE FROM orders WHERE order_id = 10004;

sink到kafka

  2. --Flink SQL
  3. Flink SQL> CREATE TABLE kafka_gmv (
  4.    day_str STRING,
  5.    gmv DECIMAL(10, 5)
  6.  ) WITH (
  7.      'connector' = 'kafka',
  8.      'topic' = 'kafka_gmv',
  9.      'scan.startup.mode' = 'earliest-offset',
  10.      'properties.bootstrap.servers' = 'localhost:9092',
  11.      'format' = 'changelog-json'
  12.  );
  14. Flink SQL> INSERT INTO kafka_gmv
  15.  SELECT DATE_FORMAT(order_date, 'yyyy-MM-dd') as day_str, SUM(price) as gmv
  16.  FROM orders
  17.  WHERE order_status = true
  18.  GROUP BY DATE_FORMAT(order_date, 'yyyy-MM-dd');
  20. -- 读取 Kafka 的 changelog 数据，观察 materialize 后的结果
  21. Flink SQL> SELECT * FROM kafka_gmv;
  • 观察kafka的数据
  2. docker-compose exec kafka bash -c 'kafka-console-consumer.sh --topic kafka_gmv --bootstrap-server kafka:9092 --from-beginning'
  • 更新 orders 数据，观察SQL CLI 和 kafka console 的输出
  2. -- 更新MySQL
  3. UPDATE orders SET order_status = true WHERE order_id = 10001;
  4. UPDATE orders SET order_status = true WHERE order_id = 10002;
  5. UPDATE orders SET order_status = true WHERE order_id = 10003;
  7. INSERT INTO orders
  8. VALUES (default, '2020-07-30 17:33:00', 'Timo', 50.00, 104, true);
  10. UPDATE orders SET price = 40.00 WHERE order_id = 10005;
  12. DELETE FROM orders WHERE order_id = 10005;

sink到hudi

  • 创建flink-hudi表
  2. Flink SQL> CREATE TABLE enriched_orders_hudi (
  3.    order_id INT PRIMARY KEY NOT ENFORCED,
  4.    order_date TIMESTAMP(3),
  5.    customer_name STRING,
  6.    price DOUBLE,
  7.    product_id INT,
  8.    order_status BOOLEAN,
  9.    `partition` VARCHAR(20)
  10.  ) PARTITIONED BY (`partition`) WITH (
  11.     'connector' = 'hudi'
  12.     ,'write.precombine.field' = 'order_date'
  13.     ,'table.type' = 'MERGE_ON_READ'
  14.     ,'path' = 'hdfs://localhost:9000/hudi/enriched_orders_hudi'
  15.     ,'compaction.tasks' = '1'
  16.         ,'compaction.trigger.strategy' = 'num_or_time'
  17.         ,'compaction.delta_commits' = '10'
  18.         ,'compaction.delta_seconds' = '10'
  19.         ,'read.tasks' = '1'
  20.         ,'read.streaming.enabled' = 'true'
  21.         ,'hoodie.datasource.query.type' = 'snapshot'
  22.         ,'read.streaming.check-interval' = '10'
  23.         ,'hoodie.datasource.merge.type' = 'payload_combine'
  24.         ,'read.utc-timezone' = 'false'
  25. );
  26. -- 插入数据
  27. Flink SQL> INSERT INTO enriched_orders_hudi
  28.     SELECT *,DATE_FORMAT(order_date, 'yyyyMMdd')
  29.     FROM orders;
  30. -- 查看表是否有数据
  31. Flink SQL> select customer_name,sum(price) as sum_price from enriched_orders_hudi group by customer_name;

spark-sql使用hudi已有数据，创建spark-sql hudi表

  • 启动spark-sql，必须用带以下参数的方式启动
  2. spark-sql --packages org.apache.hudi:hudi-spark-bundle_2.11:0.9.0,org.apache.spark:spark-avro_2.11:2.4.4 \
  3. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  4. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
  • 创建spark-sql hudi表,不指定schema，自适应，分区，主键为可选
  2. create table if not exists enriched_orders_hudi 
  3. using hudi
  4. location 'hdfs://localhost:9000/hudi/enriched_orders_hudi'
  5. options (
  6.   type = 'mor',
  7.   primaryKey = 'order_id',
  8.   preCombineField = 'order_date'
  9.  ) 
  10. partitioned by (`partition`);
  • 验证
  2. -- 在mysql的orders表中数据
  3. INSERT INTO orders
  4. VALUES 
  5. (default, default, 'A君', 500.50, 102, false),
  6. (default, default, 'B君', 100.00, 105, false);
  8. INSERT INTO orders
  9. VALUES 
  10. (default, default, 'C君', 500.50, 102, false),
  11. (default, default, 'D君', 100.00, 105, false);
  12. -- 分别在flink sql client 和spark-sql cli执行
  13. Flink SQL> select customer_name,sum(price) as sum_price from enriched_orders_hudi group by customer_name;
  14. spark-sql> select customer_name,sum(price) as sum_price from enriched_orders_hudi group by customer_name;
  15. -- 效果分别如下两图
  16. -- 更新、删除自行演示

遇到的问题

  • 由于我是单节点的hadoop，单个datanode会触发数据写入hdfs失败，报错信息如下:

Failed to replace a bad datanode on the existing pipeline due to no more good datanodes being available to try.

网上文章大部分都是让修改hdfs-site.xml添加两个配置项：

  2. <property>
  3. <name>dfs.client.block.write.replace-datanode-on-failure.enable</name>
  4. <value>true</value>
  5. </property>
  6. <property>
  7. <name>dfs.client.block.write.replace-datanode-on-failure.policy</name>
  8. <value>NEVER</value>
  9. </property>

但并不起效，因为导致这一问题的原因是客户端产生的不是服务端产生的，所以，需要修改hdfs的客户端，那么当前使用的是spark-sql，因此客户端就是spark-sql所持有的hdfs client，因此需要在启动spark-sql之后设置两个参数：

  2. spark-sql> set dfs.client.block.write.replace-datanode-on-failure.enable = ture;
  3. spark-sql> set dfs.client.block.write.replace-datanode-on-failure.policy = NEVER;

报错问题解决，并且可以通过insert into往hudi插入数据，通过select * from table查到hudi中的数据。

  • http-request java.lang.NoSuchMethodError: io.javalin.core.CachedRequestWrapper.getContentLengthLong()J

解决办法：

直接给官方提issue https://github.com/apache/hudi/issues/3797，问题解决

标签： 大数据 Flink FlinkCDC
本文转载自: https://blog.csdn.net/saqin6255/article/details/130666787
版权归原作者 白杨Shayne 所有， 如有侵权，请联系我们删除。
登录后发布评论

“使用Flink MySQL cdc分别sink到ES、Kafka、Hudi”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？