获取kafka中的数据存入mysql

注意

kafka问题

  1. 第一次我在window中安装的出现了中文乱码问题
    1. 没有解决成功，我便替换成wsl进行安装

  2. 在wsl安装之后，flink等程序连接不上kafka

    1. Timeout of 60000ms expired before the position for partition flink-cdc-0 could be determined
      1. 修改server.properties配置文件,写上你本机的IP
        1. advertised.listeners=PLAINTEXT://192.168.0.51:9092

          Flink问题

  3. 通过 sql client 进行创建任务 设置保存Checkpoint数据方式 （默认失败才会保存到磁盘，取消不会）

    1. sql client 进入命令：/bin/sql-client.sh ```shell

      第一步 设置 checkpoint 时间

      SET execution.checkpointing.interval = 3s;

第二步 默认 DELETE_ON_CANCELLATION

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/ops/state/checkpoints/#retained-checkpoints

ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint

ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint

SET ‘execution.checkpointing.externalized-checkpoint-retention’ = ‘RETAIN_ON_CANCELLATION’;

第三步 根据保存的checkpoint恢复任务

SET ‘execution.savepoint.path’ = ‘/opt/flink/ck/549543d3e61cb3bc304caa2c2d9c28f9/chk-37’;

  2.    2. java的方式实现
  3.       1. env.enableCheckpointing(3000);
  4.       2. env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  5. 2.  '**value.format' = 'csv' 问题**
  6.    1. 我使用csv格式发送kafka数据是时，mysql接收到的数据出现错误，超长，被截取，乱序等
  7.       1. 改成json就好了
  8. <a name="Rfd4s"></a>
  9. # Flink SQL准备
  10. ```sql
  11. SET execution.checkpointing.interval = 3s;
  13. CREATE TABLE kafka_source (
  14.     id int, 
  15.     name STRING,
  16.     sex STRING
  17. ) WITH (
  18.   'connector' = 'kafka',
  19.   'topic' = 'flink-cdc',
  20.   'properties.bootstrap.servers' = '192.168.0.51:9092',
  21.   'properties.group.id' = 'test-consumer-group',
  22.   'scan.startup.mode' = 'latest-offset',
  23.   'value.format' = 'json',
  24.   'value.json.fail-on-missing-field' = 'false',
  25.   'value.fields-include' = 'ALL'
  26. );
  27. // latest-offset earliest-offset
  29. -- sink
  30. CREATE TABLE mysql_sink (  
  31.      id INT,    
  32.      name STRING,     
  33.      sex STRING,
  34.      primary key (id) not enforced  
  35. ) WITH (  
  36.  'connector' = 'jdbc',  
  37.  'driver' = 'com.mysql.cj.jdbc.Driver',  
  38.  'url' = 'jdbc:mysql://192.168.0.51:3316/flink?serverTimezone=Asia/Shanghai&useSSL=false', 
  39.  'username' = 'root',  
  40.  'password' = 'root',  
  41.  'table-name' = 'mysql_sink'  
  42. );
  44. insert into mysql_sink select * from kafka_source;
  47. select * from kafka_source;
  48. select *from mysql_sink;

mysql中建表

  1. CREATE DATABASE  IF NOT EXISTS  `flink`  DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci ;
  3. CREATE TABLE `mysql_sink` (
  4.   `id` int NOT NULL AUTO_INCREMENT,
  5.   `name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  6.   `sex` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  7.   PRIMARY KEY (`id`)
  8. ) ENGINE=InnoDB AUTO_INCREMENT=908036287 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

Maven依赖

  1.     <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
  4.             <version>${flink.version}</version>
  5.         </dependency>
  7.         <!-- This dependency is provided, because it should not be packaged into the JAR file. -->
  8.         <dependency>
  9.             <groupId>org.apache.flink</groupId>
  10.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  11.             <version>${flink.version}</version>
  12.             <scope>provided</scope>
  13.         </dependency>
  14.         <dependency>
  15.             <groupId>org.apache.flink</groupId>
  16.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
  17.             <version>${flink.version}</version>
  18.             <scope>provided</scope>
  19.         </dependency>
  21.         <!-- Add connector dependencies here. They must be in the default scope (compile). -->
  23.         <!-- Example:
  25.         <dependency>
  26.             <groupId>org.apache.flink</groupId>
  27.             <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
  28.             <version>${flink.version}</version>
  29.         </dependency>
  30.         -->
  32.         <!-- Add logging framework, to produce console output when running in the IDE. -->
  33.         <!-- These dependencies are excluded from the application JAR by default. -->
  34.         <dependency>
  35.             <groupId>org.apache.logging.log4j</groupId>
  36.             <artifactId>log4j-slf4j-impl</artifactId>
  37.             <version>${log4j.version}</version>
  38.             <scope>runtime</scope>
  39.         </dependency>
  40.         <dependency>
  41.             <groupId>org.apache.logging.log4j</groupId>
  42.             <artifactId>log4j-api</artifactId>
  43.             <version>${log4j.version}</version>
  44.             <scope>runtime</scope>
  45.         </dependency>
  46.         <dependency>
  47.             <groupId>org.apache.logging.log4j</groupId>
  48.             <artifactId>log4j-core</artifactId>
  49.             <version>${log4j.version}</version>
  50.             <scope>runtime</scope>
  51.         </dependency>
  53.         <!--flink cdc -->
  54.         <dependency>
  55.             <groupId>com.ververica</groupId>
  56.             <!-- add the dependency matching your database -->
  57.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
  58.             <!-- the dependency is available only for stable releases. -->
  59.             <version>2.1.1</version>
  60.             <scope>provided</scope>
  61.         </dependency>
  62.         <!--flink cdc -->
  63.         <dependency>
  64.             <groupId>org.apache.flink</groupId>
  65.             <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
  66.             <version>${flink.version}</version>
  67.             <scope>provided</scope>
  68.         </dependency>
  69.         <dependency>
  70.             <groupId>com.google.code.gson</groupId>
  71.             <artifactId>gson</artifactId>
  72.             <version>2.8.9</version>
  73.         </dependency>
  74.         <dependency>
  75.             <groupId>org.apache.flink</groupId>
  76.             <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
  77.             <version>${flink.version}</version>
  78.             <scope>provided</scope>
  79.         </dependency>
  83.     </dependencies>

JAVA代码

  1. package cn.tannn;
  3. import org.apache.flink.streaming.api.CheckpointingMode;
  4. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.table.api.EnvironmentSettings;
  7. import org.apache.flink.table.api.SqlDialect;
  8. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  10. /**
  11.  * @author tn
  12.  * @date 2022-02-09 09:58
  13.  */
  14. public class KafkaToMySQLWitchSQL {
  15.     public static void main(String[] args) throws Exception {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         // 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】
  18.         env.enableCheckpointing(3000);
  19.         // 高级选项：
  20.         // 设置模式为exactly-once （这是默认值）
  21.         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  22.         // 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
  23.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1500);
  24.         // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
  25.         env.getCheckpointConfig().setCheckpointTimeout(60000);
  26.         // 同一时间只允许进行一个检查点
  27.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  28.         // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
  29.         //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
  30.         //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
  31.         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  32.         env.setParallelism(1);
  34.         EnvironmentSettings Settings = EnvironmentSettings.newInstance()
  35.                 .useBlinkPlanner()
  36.                 .inStreamingMode()
  37.                 .build();
  39.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, Settings);
  40.         tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
  41.             // 设置 jobName
  42.         tableEnv.getConfig().getConfiguration().setString("pipeline.name", "table_sql_job");
  44.         // latest-offset earliest-offset
  45.         String sourceDDL = "CREATE TABLE kafka_source ( " +
  46.                 "    id int,  " +
  47.                 "    name STRING, " +
  48.                 "    sex STRING " +
  49.                 ") WITH ( " +
  50.                 "  'connector' = 'kafka', " +
  51.                 "  'topic' = 'flink-cdc', " +
  52.                 "  'properties.bootstrap.servers' = '192.168.0.51:9092', " +
  53.                 "  'properties.group.id' = 'test-consumer-group', " +
  54.                 "  'scan.startup.mode' = 'latest-offset', " +
  55.                 "  'value.format' = 'json', " +
  56.                 "  'value.json.fail-on-missing-field' = 'false', " +
  57.                 "  'value.fields-include' = 'ALL' " +
  58.                 ")";
  60.         String sinkDDL = "CREATE TABLE mysql_sink (  " +
  61.                 "     id INT,    " +
  62.                 "     name STRING,     " +
  63.                 "     sex STRING, " +
  64.                 "     primary key (id) not enforced  " +
  65.                 ") WITH (  " +
  66.                 " 'connector' = 'jdbc',  " +
  67.                 " 'driver' = 'com.mysql.cj.jdbc.Driver',  " +
  68.                 " 'url' = 'jdbc:mysql://192.168.0.51:3316/flink?serverTimezone=Asia/Shanghai&useSSL=false', " +
  69.                 " 'username' = 'root',  " +
  70.                 " 'password' = 'root',  " +
  71.                 " 'table-name' = 'mysql_sink'  " +
  72.                 ")";
  74.         String transformDmlSQL = "insert into mysql_sink select * from kafka_source";
  75.         tableEnv.executeSql(sourceDDL);
  76.         tableEnv.executeSql(sinkDDL);
  77.         tableEnv.executeSql(transformDmlSQL);
  78. //        env.execute("KafkaToMySQLWitchSQL");
  79.     }
  80. }

Kafka

安装启动

  1. # STEP 1: GET KAFKA
  2. $ tar -xzf kafka_2.13-3.1.0.tgz
  3. $ cd kafka_2.13-3.1.0
  5. # STEP 2: START THE KAFKA ENVIRONMENT
  6. # > NOTE: Your local environment must have Java 8+ installed.
  7. # > Run the following commands in order to start all services in the correct order:
  8. # Start the ZooKeeper service
  9. # Note: Soon, ZooKeeper will no longer be required by Apache Kafka.
  10. $ bin/zookeeper-server-start.sh config/zookeeper.properties
  11. $ bin/windows/zookeeper-server-start.bat config/zookeeper.properties # windows
  13. # Open another terminal session and run:
  14. # Start the Kafka broker service (要等zookeeper启动好之后在执行)
  15. $ bin/kafka-server-start.sh config/server.properties
  16. $ bin/windows/kafka-server-start.bat config/server.properties # windows

创建 topic

  1. # replication-factor 指定分区的副本数
  2. # partitions 指定分区数
  3. # bootstrap-server kafka集群多个逗号隔开
  4. bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic flink-cdc
  6. bin/windows/kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic flink-cdc

查看topict详情

  1. # 查看指定的topic
  2. bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic flink-cdc
  4. bin/windows/kafka-topics.bat --describe --bootstrap-server localhost:9092 --topic flink-cdc

查看kafka topic列表

  1. bin/kafka-topics.sh --list --bootstrap-server localhost:9092
  2. bin/windows/kafka-topics.bat  --list--bootstrap-server localhost:9092

查看consumer group列表

  1. bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
  2. .\bin\windows\kafka-consumer-groups.bat --list --bootstrap-server localhost:9092
  4. # 查看详情 
  5. .\bin\windows\kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group test-consumer-group --describe

发/收数据

  1. # 生产
  2. bin/kafka-console-producer.sh --topic flink-cdc --bootstrap-server localhost:9092
  4. # 消费
  5. bin/kafka-console-consumer.sh --topic flink-cdc --from-beginning --bootstrap-server localhost:9092

Kafka命令行发送数据

  1. {"id": 1, "name":"谭宁1", "sex": "男"}
  2. {"id": 2, "name":"谭宁2", "sex": "男"}
  3. {"id": 3, "name":"谭宁3", "sex": "男"}
  4. {"id": 4, "name":"谭宁4", "sex": "男"}
  5. {"id": 5, "name":"谭宁5", "sex": "男"}
  6. {"id": 6, "name":"谭宁6", "sex": "男"}
  7. {"id": 6, "name":"谭宁6", "sex": "女"}

java 给kafka发送数据

代码

  1. package cn.tan;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.github.javafaker.Faker;
  5. import org.apache.kafka.clients.producer.KafkaProducer;
  6. import org.apache.kafka.clients.producer.ProducerConfig;
  7. import org.apache.kafka.clients.producer.ProducerRecord;
  8. import org.apache.kafka.common.serialization.StringSerializer;
  10. import java.util.HashMap;
  11. import java.util.Locale;
  12. import java.util.Map;
  13. import java.util.Properties;
  14. import java.util.concurrent.ExecutionException;
  16. /**
  17.  * 生产者
  18.  *
  19.  * @author tn
  20.  * @date 2022-02-11 09:25
  21.  */
  22. public class Producer {
  24.     private static final Faker FAKER = new Faker(Locale.CHINA);
  26.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  28.         Properties properties = new Properties();
  29.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.51:9092");
  30.         properties.put(ProducerConfig.ACKS_CONFIG,"all");
  31.         properties.put(ProducerConfig.RETRIES_CONFIG,0);
  32.         properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
  33.         properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
  34.         properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
  35.         // 序列化
  36.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  37.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  38.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  39.         /**
  40.          * topic: 消息队列的名称，可以先行在kafka服务中进行创建。如果kafka中并未创建该topic，那么便会自动创建！
  41.          * key:键值，也就是value对应的值，和Map类似\
  42.          * value:要发送的数据，数据格式为String类型的
  43.          */
  44.         while(true){
  45.             Thread.sleep(1000);
  46.             Map<String, Object> ka = new HashMap<>();
  47. //            decimalBetween(1L, 10L).intValue()
  48.             ka.put("id",FAKER.number().numberBetween(10,1000000000));
  49.             ka.put("name",FAKER.name().username());
  50.             ka.put("sex",FAKER.address().city());
  52.             String string = JSONObject.toJSONString(ka);
  53.             producer.send(new ProducerRecord<>("flink-cdc", string));
  54.             System.out.println(string);
  55.         }
  56.     }
  59. }

依赖

  1. <dependency>
  2.      <groupId>org.apache.kafka</groupId>
  3.      <artifactId>kafka-clients</artifactId>
  4.      <version>3.1.0</version>
  5. </dependency>
  6. <dependency>
  7.   <groupId>com.alibaba</groupId>
  8.   <artifactId>fastjson</artifactId>
  9.   <version>1.2.78</version>
  10. </dependency>
  11. <!-- 构造测试数据 -->
  12. <dependency>
  13.   <groupId>com.github.javafaker</groupId>
  14.   <artifactId>javafaker</artifactId>
  15.   <version>0.17.2</version>
  16. </dependency>