Flink电商指标分析项目(7)_Flink实时数据同步系统开发

今天药忘吃喽~ 2022-11-07 04:09 291阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]

## binlog日志格式分析 ##

测试日志数据

{ 
        "emptyCount": 2,
        "logFileName": "mysql-bin.000002",
        "dbName": "pyg",
        "logFileOffset": 250,
        "eventType": "INSERT",
        "columnValueList": [
            { 
                "columnName": "commodityId",
                "columnValue": "1",
                "isValid": "true"
            },
            { 
                "columnName": "commodityName",
                "columnValue": "耐克",
                "isValid": "true"
            },
            { 
                "columnName": "commodityTypeId",
                "columnValue": "1",
                "isValid": "true"
            },
            { 
                "columnName": "originalPrice",
                "columnValue": "888.0",
                "isValid": "true"
            },
            { 
                "columnName": "activityPrice",
                "columnValue": "820.0",
                "isValid": "true"
            }
        ],
        "tableName": "commodity",
        "timestamp": 1553741346000
    }

格式分析  
**字段以及说明**

<table> 
 <thead> 
  <tr> 
   <th>字段名称</th> 
   <th>说明</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>emptyCount</td> 
   <td>操作序号（第几条记录）</td> 
   <td>12</td> 
  </tr> 
  <tr> 
   <td>logFileName</td> 
   <td>binlog文件名</td> 
   <td>mysql-bin.000001</td> 
  </tr> 
  <tr> 
   <td>dbName</td> 
   <td>数据库名称</td> 
   <td>pyg</td> 
  </tr> 
  <tr> 
   <td>logFileOffset</td> 
   <td>binlog文件偏移位置</td> 
   <td>100</td> 
  </tr> 
  <tr> 
   <td>eventType</td> 
   <td>操作类型</td> 
   <td>INSERT或UPDATE或DELETE</td> 
  </tr> 
  <tr> 
   <td>columnValueList</td> 
   <td>列值列表</td> 
   <td>{ “columnName”: “列名”,“columnValue”: “列值”,“isValid”: “是否有效”}</td> 
  </tr> 
  <tr> 
   <td>tableName</td> 
   <td>表名</td> 
   <td>commodity</td> 
  </tr> 
  <tr> 
   <td>timestamp</td> 
   <td>执行时间戳</td> 
   <td>1553701139000</td> 
  </tr> 
 </tbody> 
</table>

## Flink实时同步应用开发 ##

**整体架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]  
**具体架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]

1.  Flink对接Kafka
2.  对数据进行预处理
3.  将数据落地到hbase  
    **数据同步说明**  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 3]  
    要确保hbase中的rowkey是唯一的，数据落地`不能被覆盖`

## 实时数据同步项目初始化 ##

在sync-db项目的`scala`目录中，创建以下包结构：

<table> 
 <thead> 
  <tr> 
   <th>包名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>com.itheima.syncdb.task</td> 
   <td>存放所有flink任务代码</td> 
  </tr> 
  <tr> 
   <td>com.itheima.syncdb.util</td> 
   <td>工具类</td> 
  </tr> 
  <tr> 
   <td>com.itheima.syncdb.bean</td> 
   <td>存放实体类</td> 
  </tr> 
 </tbody> 
</table>

**步骤**

1.  将pom.xml文件中的依赖导入到`sync-db`项目的pom.xml

<properties>
            <scala.version>2.11</scala.version>
            <flink.version>1.6.1</flink.version>
            <hadoop.version>2.7.5</hadoop.version>
            <hbase.version>2.0.0</hbase.version>
        </properties>
    
    
        <dependencies>
            
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_${ scala.version}</artifactId>
                <version>0.10.1.0</version>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka-0.10_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-hbase_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>${ hbase.version}</version>
            </dependency>
    
    
            
    
    
            
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>${ hadoop.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>${ hadoop.version}</version>
                
                <exclusions>
                    <exclusion>
                        <groupId>xml-apis</groupId>
                        <artifactId>xml-apis</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>${ hadoop.version}</version>
                
                <exclusions>
                    <exclusion>
                        <groupId>com.google.protobuf</groupId>
                        <artifactId>protobuf-java</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
    
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.44</version>
            </dependency>
        </dependencies>
    
    
        <build>
            <sourceDirectory>src/main/scala</sourceDirectory>
            <testSourceDirectory>src/test/scala</testSourceDirectory>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>3.0.0</version>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <artifactSet>
                                    <excludes>
                                        <exclude>com.google.code.findbugs:jsr305</exclude>
                                        <exclude>org.slf4j:*</exclude>
                                        <exclude>log4j:*</exclude>
                                    </excludes>
                                </artifactSet>
                                <filters>
                                    <filter>
                                        
                                        <artifact>*:*</artifact>
                                        <excludes>
                                            <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>com.itheima.syncdb.App</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>

1.  `sync-db`模块添加scala支持
2.  main和test创建`scala`文件夹，并标记为源代码和测试代码目录
3.  `application.conf`和`log4j.properties`配置文件

#
    #kafka的配置
    #
    # Kafka集群地址
    bootstrap.servers="hadoop102:9092,hadoop103:9092,hadoop104:9092"
    # ZooKeeper集群地址
    zookeeper.connect="hadoop102:2181,hadoop103:2181,hadoop104:2181"
    # Kafka Topic名称
    input.topic="canal"
    # 消费组ID
    group.id="canal"
    # 自动提交拉取到消费端的消息offset到kafka
    enable.auto.commit="true"
    # 自动提交offset到zookeeper的时间间隔单位（毫秒）
    auto.commit.interval.ms="5000"
    # 每次消费最新的数据
    auto.offset.reset="latest"
    
    #
    #Hbase的配置
    #
    # ZooKeeper集群地址
    hbase.zookeeper.quorum="hadoop102:2181,hadoop103:2181,hadoop104:2181"
    # HBast Master节点服务器地址
    hbase.master="hadoop102:60000"
    hbase.zookeeper.property.clientPort="2181"
    # HBase超时时长，单位（毫秒）
    hbase.rpc.timeout="600000"
    # HBase每个操作的超时时间，单位（毫秒）
    hbase.client.operator.timeout="600000"
    # HBase执行scan的超时时间，单位（毫秒）
    hbase.client.scanner.timeout.period="600000"

log4j.rootLogger=warn,stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
    log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

1.  复制之前Flink项目中的`GlobalConfigUtil`和`HBaseUtil`

## Flink程序开发 ##

**步骤**

1.  编写`App.scala`，初始化Flink环境
2.  运行Flink程序，测试是否能够消费到kafka中topic为`canal`的数据  
    **App.scala**

import com.itheima.syncdb.util.FlinkUtils
    import org.apache.flink.api.scala._
    
    object App { 
      def main(args: Array[String]): Unit = { 
    
        //获取Flink的流式环境
        val env = FlinkUtils.initFlinkStreamEnv()
    
        //整合kafka
        val consumer = FlinkUtils.initKafkaFlink()
    
        //测试打印
        val kafkaDataStream = env.addSource(consumer)
        kafkaDataStream.print()
    
    
        env.execute("SyncDBApp")
    
      }
    
    }

**FlinkUtils.scala**

package com.itheima.syncdb.util
    
    import java.util.Properties
    
    import org.apache.flink.runtime.state.filesystem.FsStateBackend
    import org.apache.flink.streaming.api.environment.CheckpointConfig
    import org.apache.flink.streaming.api.{ CheckpointingMode, TimeCharacteristic}
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
    import org.apache.flink.streaming.util.serialization.SimpleStringSchema
    
    object FlinkUtils { 
    
      // ---------------初始化FLink的流式环境--------------
      def initFlinkStreamEnv():StreamExecutionEnvironment={ 
    
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        // 设置处理的时间为EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        // 设置并行度
        env.setParallelism(1)
        // 添加Checkpoint的支持
        // 5s钟启动一次checkpoint
        env.enableCheckpointing(5000)
        // 设置checkpoint只checkpoint一次
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        // 设置两次checkpoint的最小时间间隔
        env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000)
        // 设置checkpoint的超时时长
        env.getCheckpointConfig.setCheckpointTimeout(60000)
        // 最大并行度
        env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
        // 当程序关闭时,触发额外的checkpoint
        env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
        // 设置checkpoint的地址
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-checkpoint/"))
    
        env
      }
    
      def initKafkaFlink()={ 
        // 整合Kafka
        val  props:Properties = new Properties()
    
    
        props.setProperty("bootstrap.servers",GlobalConfigUtil.bootstrapServers)
        props.setProperty("group.id",GlobalConfigUtil.groupId)
        props.setProperty("enable.auto.commit",GlobalConfigUtil.enableAutoCommit)
        props.setProperty("auto.commit.interval.ms",GlobalConfigUtil.autoCommitIntervalMs)
        props.setProperty("auto.offset.reset",GlobalConfigUtil.autoOffsetReset)
    
    
        // String topic, DeserializationSchema<T> valueDeserializer, Properties props
        val consumer = new FlinkKafkaConsumer010[String](
          GlobalConfigUtil.inputTopic,
          new SimpleStringSchema(),
          props
        )
        consumer
      }
    
    }

启动App和CanalClient 在mysql中插入一条数据

### 定义原始`Canal消息`样例类 ###

**步骤**

1.  在`bean`包下创建`Canal原始消息`映射样例类
2.  在`Cannal样例类`中编写apply方法，使用FastJSON来解析数据，转换为Cannal样例类对象
3.  编写`main`方法测试是否能够成功构建样例类对象  
    **参考代码**

import com.alibaba.fastjson.JSON
    
    
    case class Canal(var emptyCount:Long,
                     var logFileName:String,
                     var dbName:String,
                     var logFileOffset:Long,
                     var eventType:String,
                     var columnValueList:String,
                     var tableName:String,
                     var timestamp:Long)
    
    object Canal { 
      def apply(json:String): Canal = { 
        val canal = JSON.parseObject(json, classOf[Canal])
        canal
      }
    
      def main(args: Array[String]): Unit = { 
        val json = "{\"emptyCount\":1,\"logFileName\":\"mysql-bin.000022\",\"dbName\":\"pyg\",\"logFileOffset\":676,\"eventType\":\"INSERT\",\"columnValueList\":[{\"columnName\":\"commodityId\",\"columnValue\":\"1\",\"isValid\":true},{\"columnName\":\"commodityName\",\"columnValue\":\"耐克\",\"isValid\":true},{\"columnName\":\"commodityTypeId\",\"columnValue\":\"1\",\"isValid\":true},{\"columnName\":\"originalPrice\",\"columnValue\":\"888.00\",\"isValid\":true},{\"columnName\":\"activityPrice\",\"columnValue\":\"820.0\",\"isValid\":true}],\"tableName\":\"commodity\",\"timestamp\":1615432068000}"
        println(Canal(json).dbName)
    
      }
    
    }

### 解析Kafka数据流为Canal样例类 ###

**步骤**

1.  在`map`算子将消息转换为`Canal样例类`对象
2.  打印测试，如果能输出以下信息，表示成功

Canal(mysql-bin.000002,1893,pyg,commodity,DELETE,[{ "isValid":"false","columnValue":"5","columnName":"commodityId"},{ "isValid":"false","columnValue":"索菲亚","columnName":"commodityName"},{ "isValid":"false","columnValue":"3","columnName":"commodityTypeId"},{ "isValid":"false","columnValue":"35000.0","columnName":"originalPrice"},{ "isValid":"false","columnValue":"30100.0","columnName":"activityPrice"}],3,1553743567000)
    HBaseOperation(DELETE,commodity,info,5,,)

**代码:**

// 将kafka中拿到的JSON字符串canal消息，转换为Canal样例类
    val canalDataStream: DataStream[Canal] = kafkaDataStream.map { 
        json =>
        	Canal(json)
    }
        canalDataStream.print()

### 添加水印支持 ###

**步骤**

1.  使用Canal中的`timestamp`字段，生成水印数据
2.  重新运行Flink，打印添加水印后的数据

// -----------添加水印支持--------------------
       val waterDs = canalDataStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Canal] { 
    
          //当前的时间戳
          var currentTimestamp = 0L
    
          //延迟时间
          val delayTime = 2000l
    
          //返回水印的时间
          override def getCurrentWatermark: Watermark = { 
            new Watermark(currentTimestamp - delayTime)
          }
    
          // 比较当前元素的时间和上一个元素的时间,取最大值,防止时光倒流
          override def extractTimestamp(element: Canal, previousElementTimestamp: Long): Long = { 
            currentTimestamp = Math.max(element.timestamp, previousElementTimestamp)
            currentTimestamp
          }
        })
        waterDs.print()

### 定义HBaseOperation样例类 ###

HbaseOperation样例类主要封装对Hbase的操作，主要封装以下字段：

*  操作类型（opType）= INSERT/DELETE/UPDATE
 *  表名（tableName）= mysql.binlog数据库名.binlog表名
 *  列族名（cfName）= 固定为info
 *  rowkey = 唯一主键（取binlog中列数据的第一个）
 *  列名（colName）= binlog中列名
 *  列值（colValue）= binlog中列值  
    **参考代码**

/** * HBase操作样例类 * * 操作类型（opType） = INSERT/DELETE/UPDATE * 表名（tableName）= binlog数据库名_binlog表名 * 列蔟名（cfName）= 固定为info * rowkey = 唯一主键（去binlog中列数据的第一个） * 列名（colName）= binlog中列名 * 列值（colValue）= binlog中列值 */
    case class HBaseOperation(
                               var opType: String,
                               var tableName: String,
                               var cfName: String,
                               var rowkey: String,
                               var colName: String,
                               var colValue: String
                             )

### 将Canal样例类转换为HBaseOperation样例类 ###

一个binlog消息中，有会有多个列的操作。它们的映射关系如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 4]  
我们可以使用`flatMap`算子，来生成一组`HBaseOperation`操作  
**步骤**

1.  创建一个预处理任务对象
2.  使用flatMap对水印数据流转换为`HBaseOperation`
    
     *  根据eventType分别处理`HBaseOperation`列表
     *  生成的表名为`mysql.数据库名.表名`
     *  rowkey就是第一个列的值
     *  INSERT操作 -> 将所有列值转换为HBaseOperation
     *  UPDATE操作 -> 过滤掉isValid字段为`false`的列，再转换为HBaseOperation
     *  DELETE操作 -> 只生成一条DELETE的HBaseOperation的List  
        **INSERT操作记录**

<table> 
 <thead> 
  <tr> 
   <th>序号</th> 
   <th>值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>eventType</td> 
   <td>INSERT</td> 
   <td>表示这是一个插入操作</td> 
  </tr> 
  <tr> 
   <td>columnValueList</td> 
   <td>JSON数组</td> 
   <td>columnValueList中，<code>isValid</code>总是为<code>true</code></td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 5]  
**UPDATE操作记录**

<table> 
 <thead> 
  <tr> 
   <th>序号</th> 
   <th>值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>eventType</td> 
   <td>UPDATE</td> 
   <td>表示这是一个更新操作</td> 
  </tr> 
  <tr> 
   <td>columnValueList</td> 
   <td>JSON数组</td> 
   <td>columnValueList中，更新的<code>isValid</code>字段为<code>true</code>，未更新的为<code>false</code></td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 6]  
**DELETE操作记录**

<table> 
 <thead> 
  <tr> 
   <th>序号</th> 
   <th>值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>eventType</td> 
   <td>DELETE</td> 
   <td>表示这是一个删除操作</td> 
  </tr> 
  <tr> 
   <td>columnValueList</td> 
   <td>JSON数组</td> 
   <td>columnValueList中，<code>isValid</code>总是为<code>false</code></td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 7]  
**实现**

1.  在`task`包下创建`PreprocessTask`单例对象，添加`process`方法
2.  使用flatMap对Canal样例类进行扩展
3.  使用`FastJSON`解析Canal样例类中的`列值列表数据`，并存储到一个Seq中
4.  遍历集合，构建`HBaseOperation`样例类对象
5.  打印测试
6.  启动Flink验证程序是否正确处理

**JSON字符串转List**
    `List<T> parseArray(String text, Class<T> clazz) `
    `classOf[T] `: 获取class对象
    **Java的List转Scala的集合**
    注意要导入: `import scala.collection.JavaConverters._`
    `var scalaList: mutable.Buffer[T] = javaList.asScala`

**参考代码**

import java.util
    import com.alibaba.fastjson.{ JSON, JSONArray}
    import com.itheima.syncdb.bean.{ Canal, HBaseOperation}
    import org.apache.flink.streaming.api.scala.DataStream
    import org.apache.flink.api.scala._
    import scala.collection.JavaConverters._
    import scala.collection.mutable.ListBuffer
    
    /** * 封装列名、列值、是否有效 * * @param columnName 列名 * @param columnValue 列值 * @param isValid 是否有效 */
    case class ColumnValuePair(
                                var columnName: String,
                                var columnValue: String,
                                var isValid: Boolean
                              )
    
    /** * 预处理任务 * 把Canal-> HBaseOperation */
    object PreprocessTask { 
    
      def process(watermarkDataStream: DataStream[Canal]) = { 
    
        watermarkDataStream.flatMap { 
          canal => { 
            // 转换json串为集合
            val valuePairs: util.List[ColumnValuePair] = JSON.parseArray[ColumnValuePair](canal.columnValueList, classOf[ColumnValuePair])
            // java的集合转换为scala集合 需要import scala.collection.JavaConverters._
            var colValueList: mutable.Buffer[ColumnValuePair] = valuePairs.asScala
    
            // 获取操作类型
            val opType = canal.eventType
            // 生成的表名为mysql.数据库名.表名
            val tableName = s"mysql.${canal.dbName}.${canal.tableName}"
            // 列族名称
            val cfName = "info"
            // 获取mysql中的主键，获取第一列的值
            val rowkey = colValueList(0).columnValue
    
            canal.eventType match { 
              case "INSERT" =>
                // 如果是INSERT操作，将每一个列值对转换为一个HBaseOperation
                colValueList.map{ 
                  colValue =>
                    HBaseOperation(opType, tableName, cfName, rowkey, colValue.columnName, colValue.columnValue)
                }
              case "UPDATE" =>
                // UPDATE操作 -> 过滤出来isValid字段为true的列，再转换为HBaseOperation
                colValueList.filter(_.isValid).map { 
                  colValue =>
                    HBaseOperation(opType, tableName, cfName, rowkey, colValue.columnName, colValue.columnValue)
                }
              case "DELETE" =>
                // DELETE操作 -> 只生成一条DELETE的HBaseOperation的List
                List(HBaseOperation(opType, tableName, cfName, rowkey, "", ""))
            }
          }
        }
      }
    }

### Flink数据同步到hbase ###

**步骤**

1.  分两个落地实现，一个是`delete`，一个是`insert/update`（因为hbase中只有一个put操作，所以只要是insert/update都转换为put操作）
2.  启动`hbase`
3.  启动`flink`测试  
    **参考代码**

// 落地HBase 
    val operationDataStream: DataStream[HBaseOperation] = PreprocessTask.process(watermarkData)
    operationDataStream.addSink(new SinkFunction[HBaseOperation] { 
        override def invoke(value: HBaseOperation): Unit = { 
            value.opType match { 
                case "DELETE" =>
                HBaseUtil.deleteData(value.tableName, value.rowkey, value.cfName)
                case _ =>  // INSERT/UPDATE
                HBaseUtil.putData(value.tableName, value.rowkey, value.cfName, value.colName, value.colValue)
            }
        }
    })

### 验证Flink同步数据功能 ###

**步骤**

1.  启动`mysql`
2.  启动`canal`
3.  启动`zookeeper`集群
4.  启动`kafka`集群
5.  启动`hdfs`集群
6.  启动`hbase`集群
7.  启动`Flink数据同步程序`
8.  启动`Canal数据同步程序`
9.  在mysql中执行insert、update、delete语句，查看`hbase`数据是否落地

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221023/5322a998d0a24a6285ac11fe9b13476e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: /images/20221023/f8333d2bed6a40d0972b9ed9c7e12fb5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]: /images/20221023/e15c52307ed84bfcad20c0bcaf870a4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 3]: /images/20221023/cba6b3e1648849eb81b40534cb525190.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 4]: /images/20221023/09b47ebf725b47ba8148d96efba82111.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 5]: /images/20221023/a2f61c6628444dc09637b5e9f04782a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 6]: /images/20221023/d29bc43830af4b8ebb98dbbf40358a6b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 7]: /images/20221023/8a32bbefafaa4d76bee82d90f73a04cf.png