flink sql clinet 实战:模拟数据----flink-1.13.6
Posted 宝哥大数据
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了flink sql clinet 实战:模拟数据----flink-1.13.6相关的知识,希望对你有一定的参考价值。
1、模拟数据
package com.chb.flink.combat.ch1
import org.apache.kafka.clients.producer.KafkaProducer, ProducerConfig, ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util.Properties, Random
/**
* 造模拟数据
*/
object MockData2Kafka
def main(args: Array[String]): Unit =
val users = Array(1, 2, 3, 4, 5, 6)
val itemIds = Array(1001, 1002, 1003, 1004)
val categoryIds = Array(10001, 10002, 10003, 10004)
val actions = Array("pv", "buy", "cart", "fav")
val kafkaProps = new Properties()
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "chb1:9092")
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
val producer = new KafkaProducer[String, String](kafkaProps)
val topic = "user_behavior"
val random = new Random()
while (true)
val value = users(random.nextInt(users.length)) + "," + itemIds(random.nextInt(itemIds.length)) + "," +
categoryIds(random.nextInt(categoryIds.length)) + "," + actions(random.nextInt(actions.length)) +
"," + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
producer.send(new ProducerRecord[String, String](topic, value))
Thread.sleep(300)
csv格式:
user_id,item_id,category_id,event_time
4,1002,10001,buy,2022-10-17 19:25:10
6,1002,10004,buy,2022-10-17 19:25:11
5,1004,10004,cart,2022-10-17 19:25:11
1,1004,10004,buy,2022-10-17 19:25:11
5,1004,10001,fav,2022-10-17 19:25:12
5,1001,10001,pv,2022-10-17 19:25:12
1,1004,10002,fav,2022-10-17 19:25:12
1,1004,10003,cart,2022-10-17 19:25:13
2,1004,10001,buy,2022-10-17 19:25:13
以上是关于flink sql clinet 实战:模拟数据----flink-1.13.6的主要内容,如果未能解决你的问题,请参考以下文章
flink sql clinet 实战:upsert kafka connector -- flink-1.12
Flink实战系列Flink SQL 实时同步数据到 Hive
Flink 实战系列Flink SQL 实时同步 Kafka 数据到 Hudi(parquet + snappy)
Flink 实战系列Flink SQL 实时同步 Kafka 数据到 Hudi(parquet + snappy)并且自动同步数据到 Hive