Flink消费kafka消息并写入Clickhouse数据库实战

青旅半醒 2023-01-19 12:49 268阅读 0赞

**目录**

1.  kafka生产数据
2.  flink读取kafka数据并写入到clickhouse中
3.  查看clickhouse中的结果

--------------------

### kafka生产数据 ###

import java.text.SimpleDateFormat
    
    import java.util.{Date, Locale, Properties}
    
    import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
    
    import Array._
    import scala.util.Random.shuffle
    
    /**
     * @author pangsl
     * @date 2021/3/25 10:23
     */
    object KafkaProducers {
      def main(args: Array[String]): Unit = {
        SendtoKafka("test")
      }
      def SendtoKafka(topic:String): Unit = {
        val pro=new Properties()
        pro.put("bootstrap.servers", "10.50.22.21:9092")
        pro.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
        pro.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
        val producer=new KafkaProducer[String,String](pro)
        var member_id= range(1,10)
        var goods=Array("Milk","Bread","Rice","Nodles","Cookies","Fish","Meat","Fruit","Drink","Books","Clothes","Toys")
        //var ts=DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss",Locale.CHINA).format( ZonedDateTime.now())
        while (true) {
          //var ts=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())
          var ts =new SimpleDateFormat("yyyy-MM-dd").format(new Date())
          var msg = shuffle(member_id.toList).head + "\t" + shuffle(goods.toList).head + "\t" + ts + "\t" + "\n"
          print(msg)
          var record = new ProducerRecord[String, String](topic, msg)
          producer.send(record)
          Thread.sleep(8000)
        }
        //val source=Source.fromFile("C:\\UserBehavior.csv")
        //for (line<-source.getLines()){
        // val record=new ProducerRecord[String,String](topic,line)
    
        //print(ts)
        producer.close()
    
      }
    }

### flink读取kafka数据并写入到clickhouse中 ###

import java.sql.PreparedStatement
    import java.util.Properties
    import org.apache.flink.api.common.serialization.SimpleStringSchema
    import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink, JdbcStatementBuilder}
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    /**
     * @author pangsl
     * @date 2021/3/25 10:42
     */
    
    //当前版本的 flink-connector-jdbc，使用 Scala API 调用 JdbcSink 时会出现 lambda 函数的序列化问题。
    // 我们只能采用手动实现 interface 的方式来传入相关 JDBC Statement build 函数
    class CkSinkBuilder extends JdbcStatementBuilder[(Int, String, String)] {
      def accept(ps: PreparedStatement, v: (Int, String, String)): Unit = {
        ps.setInt(1, v._1)
        ps.setString(2, v._2)
        ps.setString(3, v._3)
      }
    }
    
    object To_CK {
      def main(args: Array[String]): Unit = {
    
          //获得环境
          val env = StreamExecutionEnvironment.getExecutionEnvironment
          env.setParallelism(1) //设置并发为1，防止打印控制台乱序
          env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //Flink 默认使用 ProcessingTime 处理,设置成event time
          val tEnvtEnv = StreamTableEnvironment.create(env) //Table Env 环境
          //从Kafka读取数据
          val pros = new Properties()
          pros.setProperty("bootstrap.servers", "10.50.22.21:9092")
          pros.setProperty("group.id", "test")
          pros.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
          pros.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
          pros.setProperty("auto.offset.reset", "latest")
          import org.apache.flink.api.scala._
          val dataSource = env.addSource(new FlinkKafkaConsumer011[String]("test", new SimpleStringSchema(), pros))
    
          val sql = "insert into default.testken(userid,items,create_date)values(?,?,?)"
    
          val result = dataSource.map(line => {
            val x = line.split("\t")
            println("收到数据", x(0), x(1), x(2))
    
            val member_id = x(0).trim.toLong
            val item = x(1).trim
            val times = x(2).trim
           // var time = 0l
    //        try {
    //          time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(times).getTime
    //        } //时间戳类型
    //        catch {
    //          case e: Exception => {
    //            print(e.getMessage)
    //          }
    //        }
            (member_id.toInt, item.toString, times.toString)
          }).map(x => {
            (x._1, x._2, x._3)
          }) //时间还原成datetime类型
          //result.print()
    
          result.addSink(JdbcSink.sink[(Int, String, String)](sql, new CkSinkBuilder, new JdbcExecutionOptions.Builder().withBatchSize(2).build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
              .withUrl("jdbc:clickhouse://10.50.22.21:8123")
              .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
              .withUsername("root")
              .withPassword("123456")
              .build()
          ))
    
          env.execute("To_CK")
        }
    }

### 查看clickhouse中的结果 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221021/2fd85a48eff7424983ef223d4bcd6d05.png