Flink为何而出现
人们想要高吞吐，低延迟处理数据，以前的storm只能低延迟，做不到高吞吐，spark Streaming可以高吞吐，但是更多的场景是要根据事件数据切割，或者说要实现比较复杂。
第三次学flink
2023-10-06 20:32:58
相关文章
- 10-0615分钟学 Go 第 45 天 : 使用Docker容器
- 10-06动手学深度学习64 注意力机制
- 10-06从0开始学统计-什么是中心极限定理-什么是中心极限？
- 10-06商务英语学习柯桥学外语到泓畅-老外说“go easy on me”是什么意思？
- 10-06我们来学mysql -- 查询成本之索引选择(原理篇)
- 10-06Flink + Kafka 实现通用流式数据处理详解
- 10-06胡壮麟《语言学教程》第五版PDF英文版+中文版翻译
- 10-06【flink】之kafka到kafka-2.配置Kafka数据源： Properties properties = new Properties; properties.setProperty("bootstrap.servers", "your_kafka_broker:9092"); properties.setProperty("group.id", "flink_consumer_group"); FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>( "source_topic", // Kafka source topic new SimpleStringSchema, // 数据反序列化方式 properties ); DataStream<String> kafkaStream = env.addSource(kafkaConsumer); 3.数据处理（可选）： DataStream<String> processedStream = kafkaStream.map(value -> value.toUpperCase); 4.配置Kafka数据目标： FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>( "target_topic", // Kafka target topic new SimpleStringSchema, // 数据序列化方式 properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE_SEMANTICS // 确保数据精确一次处理（可选） ); 5.将数据写入Kafka： processedStream.addSink(kafkaProducer); 6.启动Flink作业： 将上述代码整合到一个Java类中，并在main方法中启动Flink执行环境： public class FlinkKafkaToKafka { public static void main(String args) throws Exception { // 创建Flink执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment; env.setParallelism(1); // 配置Kafka数据源 Properties properties = new Properties; properties.setProperty("bootstrap.servers", "your_kafka_broker:9092"); properties.setProperty("group.id", "flink_consumer_group"); FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>( "source_topic", new SimpleStringSchema, properties ); DataStream<String> kafkaStream = env.addSource(kafkaConsumer); // 数据处理（可选） DataStream<String> processedStream = kafkaStream.map(value -> value.toUpperCase); // 配置Kafka数据目标 FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>( "target_topic", new SimpleStringSchema, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE_SEMANTICS ); // 将数据写入Kafka processedStream.addSink(kafkaProducer); // 启动Flink作业 env.execute("Flink Kafka to Kafka Job"); } } 五、运行与验证
- 10-06Flink CDC系列之：理解学习Kubernetes模式-提交 Flink CDC Job
- 10-06从头学PHP之数组输出&基本函数