Apache Flink：流处理革命的领导者与新一代大数据计算引擎

MemorySize taskHeapSize = MemorySize.ofMebiBytes(1024); // 任务堆内存 MemorySize managedMemorySize = MemorySize.ofMebiBytes(512); // 托管内存 MemorySize networkBufferSize = MemorySize.ofMebiBytes(128); // 网络缓冲区

// 序列化机制优化 TypeInformation typeInfo = BasicTypeInfo.STRING_TYPE_INFO; Serializer serializer = typeInfo.createSerializer(executionConfig);

public class RealTimeProcessingJob { public static void main(String[] args) throws Exception { // 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置检查点间隔（开启精确一次语义） env.enableCheckpointing(5000); // 5秒一次

// 定义数据源（从Kafka读取） DataStream stream = env .addSource(new FlinkKafkaConsumer<>( "input-topic", new SimpleStringSchema(), properties)) .name("kafka-source");

// 数据转换处理 DataStream> processed = stream .flatMap(new Tokenizer()) // 分词 .keyBy(value -> value.f0) // 按单词分组 .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) // 10秒滚动窗口 .sum(1) // 求和 .name("word-count");

// 输出到外部系统 processed.addSink(new FlinkKafkaProducer<>( "output-topic", new SimpleStringSchema(), properties)) .name("kafka-sink");

// 执行作业 env.execute("Real-time Word Count"); }

// 分词器实现 public static class Tokenizer extends RichFlatMapFunction> {

@Override public void flatMap(String value, Collector> out) { String[] words = value.toLowerCase().split("\\W+"); for (String word : words) { if (!word.isEmpty()) { out.collect(new Tuple2<>(word, 1)); } } } } }

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 创建数据源表 tableEnv.executeSql( "CREATE TABLE user_actions (" + " user_id STRING, " + " action_type STRING, " + " action_time TIMESTAMP(3), " + " WATERMARK FOR action_time AS action_time - INTERVAL '5' SECOND" + ") WITH (" + " 'connector' = 'kafka'," + " 'topic' = 'user-actions'," + " 'properties.bootstrap.servers' = 'localhost:9092'," + " 'format' = 'json'" + ")" );

// 执行SQL查询 Table result = tableEnv.sqlQuery( "SELECT " + " user_id, " + " COUNT(*) as action_count, " + " TUMBLE_END(action_time, INTERVAL '1' HOUR) as window_end " + "FROM user_actions " + "GROUP BY user_id, TUMBLE(action_time, INTERVAL '1' HOUR)" );

// 转换为DataStream输出 DataStream resultStream = tableEnv.toDataStream(result, Result.class); resultStream.print();

DataStream withTimestampsAndWatermarks = stream .assignTimestampsAndWatermarks( WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) -> event.getCreationTime()) );

DataStream sensorData = ...;

// 滚动窗口：每5分钟统计一次 DataStream tumblingWindow = sensorData .keyBy(SensorReading::getSensorId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .aggregate(new AverageAggregate());

// 滑动窗口：每1分钟输出过去5分钟的平均值 DataStream slidingWindow = sensorData .keyBy(SensorReading::getSensorId) .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))) .aggregate(new AverageAggregate());

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每5秒生成一个检查点 env.enableCheckpointing(5000);

// 精确一次语义 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 检查点超时时间 env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同时保留的检查点数量 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

wget https://archive.apache.org/dist/flink/flink-1.16.0/flink-1.16.0-bin-scala_2.12.tgz tar -xzf flink-1.16.0-bin-scala_2.12.tgz cd flink-1.16.0

# 启动集群 ./bin/start-cluster.sh

# 提交作业 ./bin/flink run ./examples/streaming/WordCount.jar

./bin/flink run -m yarn-cluster -yn 2 -ys 1024 -yjm 1024 ./examples/streaming/WordCount.jar

# Per-Job模式（已弃用，推荐Application模式） ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=1024m \ ./examples/streaming/WordCount.jar

kind: FlinkDeployment metadata: name: word-count spec: image: flink:1.16.0 flinkVersion: v1_16 image: flink:1.16.0-scala_2.12 serviceAccount: flink jobManager: resource: memory: "2048Mi" cpu: 1 taskManager: resource: memory: "2048Mi" cpu: 1 replicas: 2 job: jarURI: local:///opt/flink/examples/streaming/WordCount.jar parallelism: 2

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 并行度设置 env.setParallelism(4);

// 缓冲区超时（吞吐量与延迟的权衡） env.setBufferTimeout(100);

// 对象重用模式（减少序列化开销） env.getConfig().enableObjectReuse();

// 时间特性（使用事件时间） env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

public class RealTimeETLJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 从Kafka读取原始数据 DataStream rawData = env.addSource( new FlinkKafkaConsumer<>("raw-data", new SimpleStringSchema(), props));

// 数据清洗和转换 DataStream cleanData = rawData .filter(new DataQualityFilter()) // 数据质量过滤 .map(new DataTransformer()) // 数据转换 .keyBy(CleanData::getUserId) .process(new Deduplication()) // 数据去重 .name("data-cleansing");

// 输出到多个目标系统 cleanData.addSink(new ElasticsearchSink<>(esSinkConfig)); // 到Elasticsearch cleanData.addSink(new JDBCSink(jdbcUrl, username, password)); // 到关系数据库

env.execute("Real-time ETL Pipeline"); } }

public class FraudDetectionJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream transactions = ...;

DataStream alerts = transactions .keyBy(Transaction::getUserId) .process(new FraudDetectionPattern()) .name("fraud-detection");

// 复杂模式：检测短时间内多次大额交易 public static class FraudDetectionPattern extends KeyedProcessFunction {

private transient ValueState lastTransactionTimeState; private transient ValueState totalAmountState;

@Override public void open(Configuration parameters) { // 状态初始化 }

@Override public void processElement(Transaction transaction, Context ctx, Collector out) throws Exception { // 风控逻辑实现 if (isSuspiciousPattern(transaction)) { out.collect(new Alert(transaction, "SUSPICIOUS_PATTERN")); } } } } }

Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "flink-consumer"); properties.setProperty("isolation.level", "read_committed");

FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>( "input-topic", new SimpleStringSchema(), properties);

// 从最早位置开始读取（故障恢复时） consumer.setStartFromEarliest();

String name = "myhive"; String defaultDatabase = "mydatabase"; String hiveConfDir = "/opt/hive-conf";

HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir); tableEnv.registerCatalog("myhive", hive); tableEnv.useCatalog("myhive");