大数据之使用Flink消费Kafka中topic为ods_mall_log的数据，根据不同的表前缀区分在存入Kafka的topic当中-Toy模板网

这篇具有很好参考价值的文章主要介绍了大数据之使用Flink消费Kafka中topic为ods_mall_log的数据，根据不同的表前缀区分在存入Kafka的topic当中。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

前言

本题来源于全国职业技能大赛之大数据技术赛项赛题 - 电商数据处理 - 实时数据处理

注：由于设备问题，代码执行结果以及数据的展示无法给出，可参照我以往的博客其中有相同数据源展示

题目：

使用Flink消费Kafka中topic为ods_mall_log的数据，根据数据中不同的表前缀区分，将数据分别分发至kafka的DWD层的dim_customer_login_log的Topic中，其他的表则无需处理；

提示：以下是本篇文章正文内容，下面案例可供参考（使用Scala语言编写）

一、读题分析

涉及组件：Scala，Flink，Kafka，HBase

涉及知识点：

Flink函数的使用
了解HBase，基本使用HBase

二、处理过程

1.数据处理部分：


import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}

import java.util.Properties

object answer2 {
  def main(args: Array[String]): Unit = {
    import moduleC.test.HBaseSink2
    import org.apache.flink.streaming.api.scala.DataStream

    //    创建flink流环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //    创建Kafka的配置
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "bigdata1:9092")

    //    创建Kafka的消费者
    val kafkaConsumer = new FlinkKafkaConsumer[String]("ods_mall_log", new SimpleStringSchema(), properties)

    //    读取消费的数据
    val kafkaStream = env.addSource(kafkaConsumer)

    val newStream: DataStream[(String, String)] = kafkaStream
      .map(
        line => {
          val tablename = line.split(":")(0)
          val data = line.split(":")(1).stripPrefix("(").stripSuffix(");")
          //        12115|7611|0|0|'20230407111600'
          (tablename, data)
        }
      ).filter(_._1 == "customer_login_log")

    newStream.print()
    //    创建Kafka的生产者
    val dwdKafkaProduce = new FlinkKafkaProducer[String]("dim_customer_login_log", new SimpleStringSchema(), properties)
    //    val dwdKafkaProduce2 = new FlinkKafkaProducer[String]("dwd-topic1", new SimpleStringSchema(), properties)

    //    向Kafka发送数据
    //    newStream.addSink(dwdKafkaProduce)    不调用map方法会报错因为需要一个string而不是（string，string）
    val result = newStream.map(line => line._2)
    result.addSink(dwdKafkaProduce)

    //    发到HBase上
    result.addSink(new HBaseSink2)

    //    execute
    env.execute("flinkKafkaToKafka")
  }
}

2.HBaseSink（未经测试，不能证明其正确性，仅供参考！）

package moduleC.test

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.client
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.ConnectionFactory

class HBaseSink(tableName: String, columnFamily: String) extends RichSinkFunction[String] {

   private var connection: client.Connection = _
   private var table: client.Table = _

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    // 创建 HBase 连接
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "bigdata1")
    conf.set("hbase.zookeeper.property.clientPort", "2181")
    connection = ConnectionFactory.createConnection(conf)
    // HBase 表信息
    table = connection.getTable(TableName.valueOf(tableName))
  }

  override def invoke(value: String, context: SinkFunction.Context): Unit = {
    val put = new Put(Bytes.toBytes(value))
    put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("data"), Bytes.toBytes(value))
    table.put(put)
  }

  override def close(): Unit = {
    if (table != null) table.close()
    if (connection != null) connection.close()
  }

}