大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载本文以 Apache Flink 仓库中 PyFlink DataStream 参考文档 为主线系统梳理 PyFlink DataStream API 的完整公开接口从StreamExecutionEnvironment的作业构建与执行模式到DataStream的转换算子、状态与定时器、窗口与检查点、侧输出再到文件/Kafka/Pulsar/JDBC 等连接器与 Avro/CSV/JSON/Parquet 格式。读完本文你将掌握 PyFlink DataStream 编程模型的全貌并能在开发时按模块快速定位到具体的类与方法配合源码理解底层行为。文档定位PyFlink DataStream API 的官方索引flink-python/docs/reference/pyflink.datastream/目录是 PyFlink DataStream API 的官方参考文档集合其中 index.rst 是总入口页声明了该页面的定位——This page gives an overview of all public PyFlink DataStream API即覆盖所有公开的 PyFlink DataStream API。它通过 Sphinxtoctree指令组织出 10 个主题子文档子文档主题stream_execution_environment.rst执行环境与作业控制datastream.rst数据流核心类型与算子functions.rst用户自定义函数state.rst状态与状态后端timer.rst定时服务与时间域window.rst窗口、触发器与窗口分配器checkpoint.rst检查点配置与存储sideoutput.rst侧输出标签connectors.rst各类外部系统连接器formats.rst序列化格式下面按编程模型的生命周期顺序展开先构建执行环境再定义数据流算子最后接入状态、时间与外部系统。StreamExecutionEnvironment流式程序的执行上下文stream_execution_environment.rst 定义StreamExecutionEnvironment为流式程序被执行的上下文LocalStreamEnvironment会在附着的 JVM 内执行RemoteStreamEnvironment会在远程集群上执行。它提供了两类能力控制作业执行设置并行度、容错/检查点参数、缓冲超时、算子链、重启策略等与外部世界交互接入数据源、注册 Python 依赖文件、requirements、归档、解释器、注册缓存文件等。该环境暴露的方法覆盖了作业生命周期的每个阶段按其职责可归纳为并行度与资源set_parallelism/get_parallelism、set_max_parallelism/get_max_parallelism、set_default_local_parallelism/get_default_local_parallelism、register_slot_sharing_group配合SlotSharingGroup与MemorySize使用执行模式与算子链set_runtime_mode、set_buffer_timeout/get_buffer_timeout、disable_operator_chaining/is_chaining_enabled容错与状态enable_checkpointing/get_checkpoint_interval/get_checkpointing_mode、get_checkpoint_config、set_state_backend/get_state_backend、enable_changelog_state_backend/is_changelog_state_backend_enabled、set_default_savepoint_directory/get_default_savepoint_directory、set_restart_strategy/get_restart_strategy、is_unaligned_checkpoints_enabled/is_force_unaligned_checkpoints类型与序列化add_default_kryo_serializer、register_type_with_kryo_serializer、register_type、set_stream_time_characteristic/get_stream_time_characteristic、configurePython 依赖注入add_python_file、set_python_requirements、add_python_archive、set_python_executable、add_jars、add_classpaths数据接入create_input、add_source、from_source、read_text_file、from_collection、register_cached_file作业提交execute、execute_async、get_execution_plan、close以及便捷的get_execution_environment静态工厂。其中enable_checkpointing、set_state_backend、set_restart_strategy等是每个生产作业的标准动作与后文 Checkpoint 小节 的配置项一一对应。RuntimeExecutionMode流批一体的执行模式文档专门解释了RuntimeExecutionMode——它控制 DataStream 程序的运行时执行模式影响任务调度、网络 Shuffle 行为和时间语义部分算子也会依据执行模式改变记录发射行为。三种取值STREAMING流水线以流式语义执行。所有任务在执行开始前全部部署启用检查点处理时间和事件时间都被完整支持BATCH流水线以批式语义执行。任务按所属调度区域渐进式调度区域间的 Shuffle 是阻塞式的watermark 被假定为完美即不存在迟到数据处理时间在执行期间被假定为不推进AUTOMATICFlink 自动判定——若所有 Source 都是有界数据源则采用 BATCH若存在至少一个无界数据源则采用 STREAMING。这一枚举与 Java 侧RuntimeExecutionMode完全对齐在 PyFlink 中通过StreamExecutionEnvironment.set_runtime_mode(...)设置是流批一体能力在 DataStream API 上的直接体现。DataStream 核心类型与算子datastream.rst 是篇幅最大的子文档它定义了 PyFlink DataStream API 的核心类型体系其中DataStream被定义为同类型元素的流a stream of elements of the same type。其源码实现位于 pyflink/datastream/data_stream.py。DataStream基础流类型DataStream的方法按用途分为三类元信息与算子属性对应 Java DataStream 的算子描述能力get_name/name算子名、uid/set_uid_hash算子 UID用于状态恢复时的稳定标识、set_parallelism/set_max_parallelism、get_type、get_execution_environment、force_non_parallel、set_buffer_timeout、start_new_chain/disable_chaining算子链控制、slot_sharing_group、set_description。转换算子Transformationmap、flat_map、key_by、filter、window_all、union、connect、project、process、assign_timestamps_and_watermarks、add_sink、sink_to。这些算子构成流式处理的主体例如key_bydata_stream.py#L357接收一个 Callable 或KeySelector将流按 key 分组返回KeyedStreamwindow_alldata_stream.py#L457接收WindowAssigner返回AllWindowedStreamassign_timestamps_and_watermarksdata_stream.py#L666接收WatermarkStrategy注入 watermark 生成逻辑。数据重分区Partitioningshuffle、rescale、rebalance、forward、broadcast、partition_custom。这些算子只改变数据在并行子任务间的分发策略不改变业务逻辑是调整数据倾斜、控制网络传输方式的关键手段。结果输出与旁路execute_and_collectdata_stream.py#L839支持job_execution_name与limit参数用于本地调试时同步收集结果、print、get_side_output、cache[data_stream.py#L899](https://link.gitcode.com/i/ba53962e20fcc4cc9b0bc95b26a87f11#L899)返回CachedDataStream。KeyedStream按键分组的流KeyedStream由key_by产生文档描述为按 key 分组的流。它继承DataStream的全部能力并新增了针对每个 key 独立计算的算子reduce、sum、min、max、min_by、max_by聚合、window/count_window窗口化、processKeyedProcessFunction可访问 keyed state 与定时器。由于数据已按 key 分区这些算子的状态和窗口计算天然具备 per-key 隔离性。WindowedStream 与 AllWindowedStream窗口流WindowedStream表示元素按 key 分组每个 key 的流再依据WindowAssigner切分为窗口的流。文档特别强调两点窗口按 key 分别求值每个 key 的窗口可以在不同时刻触发WindowedStream纯粹是 API 层面的构造运行时会被折叠collapse成 KeyedStream 与窗口操作的一个单一算子。其方法包括trigger自定义触发器、allowed_lateness允许的迟到时间、side_output_late_data迟到数据侧输出、reduce、aggregate、apply、process。AllWindowedStream表示不按键分组的窗口流全局窗口化。文档额外说明若指定了Evictor它会在 Trigger 触发之后、窗口实际求值之前驱逐窗口内元素使用 evictor 会显著降低窗口性能因为无法使用窗口结果的预聚合。其方法集与WindowedStream一致trigger、allowed_lateness、side_output_late_data、reduce、aggregate、apply、process运行时会与窗口操作折叠为单一算子。ConnectedStreams、BroadcastStream 与 BroadcastConnectedStream多流关联ConnectedStreams表示两条可能类型不同流的连接常用于一条流上的操作直接受另一条流影响的场景典型示例是将随时间变化的规则流作用到数据流上一条流持有规则另一条流持有待应用规则的元素连接后的算子把当前规则集维护在状态中收到规则更新则更新状态收到数据元素则用状态中的规则处理它。概念上可看作Either类型的并集流。方法key_by、map、flat_map、process。BroadcastStream用于广播状态的分发BroadcastConnectedStream表示将 keyed 或非 keyed 流与携带BroadcastState的BroadcastStream连接的结果。文档给出的典型场景同样是动态规则广播流持有规则并存入 broadcast state从而在所有并行实例上可用可以应用到另一条流的全部分区。其核心方法是process配合BroadcastProcessFunction/KeyedBroadcastProcessFunction。DataStreamSink 与 CachedDataStreamDataStreamSink是用于从流式拓扑发射元素的流式 Sink提供name、uid、set_uid_hash、set_parallelism、set_description、disable_chaining、slot_sharing_group等算子属性设置方法CachedDataStream表示中间结果在首次计算时被缓存的流后续作业使用同一个CachedDataStream可复用缓存结果、避免重复计算data_stream.py#L1774 的cache方法产生data_stream.py#L1788 的invalidate方法使缓存失效。它提供与DataStream几乎等价的算子集map、flat_map、key_by、filter、window_all、union、connect、各类重分区、process、assign_timestamps_and_watermarks、add_sink、sink_to、execute_and_collect、print、get_side_output等并额外提供invalidate。Functions用户自定义函数族functions.rst 罗列了 DataStream API 中全部用户自定义函数All user-defined functions以及运行时上下文。RuntimeContext是函数访问运行时信息的入口提供任务与子任务信息get_task_name、get_number_of_parallel_subtasks、get_max_number_of_parallel_subtasks、get_index_of_this_subtask、get_attempt_number、get_task_name_with_subtasks、作业参数get_job_parameter、指标组get_metrics_group以及获取各类状态的方法get_stateValueState、get_list_state、get_map_state、get_reducing_state、get_aggregating_state。函数接口完整清单如下与DataStream/KeyedStream的算子一一对应基础单流函数MapFunction、FlatMapFunction、FilterFunction、ReduceFunction、AggregateFunction、ProcessFunction、KeyedProcessFunction双流连接函数CoMapFunction、CoFlatMapFunction、CoProcessFunction、KeyedCoProcessFunction窗口函数WindowFunction、AllWindowFunction、ProcessWindowFunction、ProcessAllWindowFunction广播函数BroadcastProcessFunction、KeyedBroadcastProcessFunction辅助接口KeySelectorkey 提取、NullByteKeySelector、Partitioner配合partition_custom。其中ProcessFunction系列是最灵活的底层抽象它们能访问RuntimeContext、维护状态、注册定时器并向侧输出发射数据是map/filter等简单函数无法覆盖复杂场景时的首选。State状态类型、TTL 与状态后端state.rst 围绕状态展开三个层次状态类型、状态描述符与 TTL 配置、状态后端。状态类型与描述符状态存储接口OperatorStateStore.get_broadcast_state获取广播状态状态值类型ValueState单值、AppendingState、MergingState、ReducingState归约值、AggregatingState聚合值、ListState列表、MapState映射、ReadOnlyBroadcastState/BroadcastState广播状态状态描述符ValueStateDescriptor、ListStateDescriptor、MapStateDescriptor、ReducingStateDescriptor、AggregatingStateDescriptor——描述符声明了状态的名字、类型与默认值是RuntimeContext.get_state(...)等方法的入参。StateTtlConfig状态过期策略StateTtlConfig是状态 TTLTime To Live的配置载体文档通过autoclass展开了四个枚举/类UpdateType状态过期时间的更新策略创建与写入时更新update_ttl_on_create_and_write或读写时都更新update_ttl_on_read_and_writeStateVisibility过期状态对读的可见性return_expired_if_not_cleaned_up返回未清理的过期数据never_return_expired永不返回过期数据TtlTimeCharacteristicTTL 计时特征默认使用处理时间use_processing_timeCleanupStrategies过期数据清理策略——cleanup_full_snapshot全量快照清理、cleanup_incrementally增量清理、cleanup_in_rocksdb_compact_filterRocksDB compaction filter 清理、disable_cleanup_in_background禁用后台清理。配置方式遵循 Builder 模式new_builder创建构建器通过set_update_type、set_state_visibility、set_ttl_time_characteristic、set_ttl等设置最终build产出查询方法包括get_update_type、get_state_visibility、get_ttl、get_ttl_time_characteristic、is_enabled、get_cleanup_strategies。StateBackend状态存储的底层机制文档对StateBackend给出了较完整的原理阐述定义状态后端决定流式应用的状态在集群内如何本地存储。HashMapStateBackend把工作状态保存在 TaskManager 内存中轻量且无额外依赖EmbeddedRocksDBStateBackend把工作状态保存在内嵌 RocksDB 实例中可扩展到 TB 级仅受所有 TaskManager 可用磁盘空间限制原始字节存储与后端StateBackend为原始字节存储raw bytes storage以及 keyed state、operator state 创建服务。由它创建的AbstractKeyedStateBackend与OperatorStateBackend决定 key 与算子的工作状态如何持有、如何通过CheckpointStreamFactory做检查点可序列化性状态后端需要实现java.io.Serializable因为要与流式应用代码一起分发到并行进程因此实现被设计为轻量工厂只含配置由工厂创建真正的状态存储线程安全多个线程可能并发创建流与 keyed/operator 状态后端因此实现必须线程安全。PyFlink 暴露的后端类包括HashMapStateBackend、EmbeddedRocksDBStateBackend以及为兼容旧版本保留的MemoryStateBackend、FsStateBackend、RocksDBStateBackend、CustomStateBackend自定义后端和PredefinedOptions预定义 RocksDB 选项。Timer定时服务与时间域timer.rst 定义了时间相关的三类 APITimerService定时服务提供current_processing_time当前处理时间、current_watermark当前 watermark、register_processing_time_timer注册处理时间定时器、register_event_time_timer注册事件时间定时器、delete_processing_time_timer/delete_event_time_timer删除对应定时器。定时器通常通过KeyedProcessFunction/ProcessFunction的上下文获取是事件时间迟到处理、会话超时检测等场景的基石TimeCharacteristic时间特征对应流处理的时间语义设置TimeDomain时间域区分处理时间ProcessingTime与事件时间EventTime两种定时器注册域。Window窗口、触发器与窗口分配器window.rst 完整覆盖窗口机制的三大构件Window 类型TimeWindow时间窗口含起止时间戳、CountWindow计数窗口、GlobalWindow全局窗口永不自然触发Trigger触发器决定窗口何时求值/清理包括TriggerResult触发结果枚举、EventTimeTrigger/ContinuousEventTimeTrigger、ProcessingTimeTrigger/ContinuousProcessingTimeTrigger、PurgingTrigger清理型触发器、CountTrigger、NeverTrigger永不触发WindowAssigner窗口分配器决定元素进入哪个窗口包括MergingWindowAssigner可合并窗口分配器的基类、计数类CountTumblingWindowAssigner/CountSlidingWindowAssigner、时间类TumblingProcessingTimeWindows/TumblingEventTimeWindows/SlidingProcessingTimeWindows/SlidingEventTimeWindows、会话类ProcessingTimeSessionWindows/EventTimeSessionWindows/DynamicProcessingTimeSessionWindows/DynamicEventTimeSessionWindows、以及GlobalWindowsSessionWindowTimeGapExtractor为动态会话窗口提供会话间隙提取器可从元素中动态计算会话时间间隙。这些类与KeyedStream.window(...)/DataStream.window_all(...)配合使用例如滚动窗口TumblingEventTimeWindows.of(Time.minutes(5))、滑动窗口SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(1))、会话窗口EventTimeSessionWindows.with_gap(Time.minutes(5))。Checkpoint检查点配置与 CheckpointStoragecheckpoint.rst 定义检查点相关的配置与存储机制。CheckpointConfig 默认值与配置方法CheckpointConfig是捕获所有检查点相关设置的配置文档明确给出四个默认常量常量含义默认值DEFAULT_MODE默认检查点模式精确一次exactly onceDEFAULT_TIMEOUT单次检查点尝试的超时10 分钟DEFAULT_MIN_PAUSE_BETWEEN_CHECKPOINTS两次检查点之间的最小暂停无noneDEFAULT_MAX_CONCURRENT_CHECKPOINTS并发检查点数量上限1配置方法覆盖完整生命周期开关与模式is_checkpointing_enabled、get/set_checkpointing_mode、get/set_checkpoint_interval、超时与节奏get/set_checkpoint_timeout、get/set_min_pause_between_checkpoints、get/set_max_concurrent_checkpoints、错误容忍is/set_fail_on_checkpointing_errors、get/set_tolerable_checkpoint_failure_number、外部化检查点enable_externalized_checkpoints、set_externalized_checkpoint_cleanup、is_externalized_checkpoints_enabled、get_externalized_checkpoint_cleanup配合ExternalizedCheckpointCleanup/ExternalizedCheckpointRetention枚举、非对齐检查点is_unaligned_checkpoints_enabled、enable/disable_unaligned_checkpoints、set/get_alignment_timeout、set/is_force_unaligned_checkpoints、存储设置set/get_checkpoint_storage、set_checkpoint_storage_dir。CheckpointStorage检查点持久化策略文档说明检查点存储定义了StateBackend如何为流式应用存储容错状态JobManagerCheckpointStorage把检查点保存在 JobManager 内存中轻量、无额外依赖但不可扩展、仅支持小状态适合本地测试与开发FileSystemCheckpointStorage把检查点保存在文件系统HDFS、NFS、S3、GCS 等支持数 TB 级大状态并为流式应用提供高可用基础推荐用于大多数生产部署。与 StateBackend 类似CheckpointStorage也遵循轻量工厂设计实现需可序列化java.io.Serializable以随应用代码分发需线程安全以支持并发创建流原始字节存储服务通过CheckpointStreamFactory被 JobManager 用于存储检查点与恢复元数据也通常被 keyed/operator 状态后端用于存储检查点状态。PyFlink 暴露JobManagerCheckpointStorage、FileSystemCheckpointStorage与CustomCheckpointStorage。Side Outputs侧输出标签sideoutput.rst 讲解侧输出机制的核心——OutputTag它被定义为用于标记算子侧输出的、带类型与名字的标签。文档给出了可直接运行的示例展示了 PyFlink 的三种构造方式与一个约束# 显式指定输出类型 info OutputTag(late-data, Types.TUPLE([Types.STRING(), Types.LONG()])) # 隐式将 list 包装为 Types.ROW info_row OutputTag(row, [Types.STRING(), Types.LONG()]) # 隐式使用 pickle 序列化 info_side OutputTag(side) # ERROR: tag id 不能为空字符串Python API 的额外要求 info_error OutputTag()注意最后一行展示的额外约束PyFlink 要求 tag id 不能为空字符串这是 Python API 特有的校验Java API 无此限制。侧输出典型用于处理迟到数据在窗口算子中通过side_output_late_data(tag)把迟到元素发射到侧输出流再通过DataStream.get_side_output(tag)取回单独处理。Connectors外部系统连接器connectors.rst 覆盖 PyFlink DataStream API 支持的全部连接器是接入外部数据源/目标的清单File System文件 SourceFileSource/FileSourceBuilder配套FileEnumeratorProvider、FileSplitAssignerProvider、StreamFormat、BulkFormat文件 SinkFileSink、StreamingFileSink配套BucketAssigner分桶策略、RollingPolicy/DefaultRollingPolicy/OnCheckpointRollingPolicy滚动策略、OutputFileConfig、FileCompactStrategy/FileCompactor文件压缩Number SequenceNumberSequenceSource用于生成有界数字序列常见于压测与测试场景Kafka两代 API 并存——旧版FlinkKafkaConsumer/FlinkKafkaProducer含Semantic语义枚举与新版KafkaSource/KafkaSourceBuilder配套KafkaTopicPartition、KafkaOffsetResetStrategy、KafkaOffsetsInitializer以及KafkaSink/KafkaSinkBuilder配套KafkaRecordSerializationSchema/KafkaRecordSerializationSchemaBuilder、KafkaTopicSelectorKinesisSourceFlinkKinesisConsumer配套KinesisShardAssigner、KinesisDeserializationSchema、WatermarkTrackerSinkKinesisStreamsSink/KinesisStreamsSinkBuilder、KinesisFirehoseSink/KinesisFirehoseSinkBuilder配套PartitionKeyGeneratorPulsarSourcePulsarSource/PulsarSourceBuilder配套StartCursor、StopCursor、RangeGeneratorSinkPulsarSink/PulsarSinkBuilder配套TopicRoutingMode、MessageDelayerJDBCJdbcSink配套JdbcConnectionOptions连接配置与JdbcExecutionOptions执行配置RabbitMQRMQRMQSource/RMQSink配套RMQConnectionConfigElasticsearchElasticsearchSink配套Elasticsearch6SinkBuilder/Elasticsearch7SinkBuilder、ElasticsearchEmitter、FlushBackoffTypeflush 退避策略CassandraCassandraSink配套ConsistencyLevel、MapperOptions、ClusterBuilder、CassandraCommitter、CassandraFailureHandler。Formats序列化格式formats.rst 定义 DataStream API 可用的数据格式AvroAvroSchema、GenericRecordAvroTypeInfo、AvroInputFormat、AvroBulkWriters、AvroRowDeserializationSchema/AvroRowSerializationSchemaCSVCsvSchema/CsvSchemaBuilder、CsvReaderFormat、CsvBulkWriters、CsvRowDeserializationSchema/CsvRowSerializationSchemaJSONJsonRowDeserializationSchema/JsonRowSerializationSchemaORCOrcBulkWriters列式批量写入ParquetAvroParquetReaders/AvroParquetWriters、ParquetColumnarRowInputFormat、ParquetBulkWriters。其中*RowDeserializationSchema/*RowSerializationSchema系列用于连接器如 Kafka的 record 编解码*BulkWriters系列用于FileSink的列式/批量写出。实践要点如何基于该参考定位开发综合上述参考文档与源码给出几条可直接落地的使用建议按编程模型顺序组织代码StreamExecutionEnvironment.get_execution_environment()→ 通过from_source/from_collection/read_text_file接入数据 → 用key_by/map/process等转换 → 用window/assign_timestamps_and_watermarks处理时间与窗口 → 用sink_to/add_sink输出 → 最后execute提交。每个环节的 API 都能在本参考的对应子文档中找到。状态与检查点配套使用set_state_backend如HashMapStateBackend/EmbeddedRocksDBStateBackend与set_checkpoint_storage如FileSystemCheckpointStorage分别决定工作状态与检查点落盘方式生产环境推荐后者TTL 需求用StateTtlConfig.new_builder()链式配置后传入StateDescriptor。时间语义先行事件时间处理必须先通过assign_timestamps_and_watermarks注入WatermarkStrategy窗口才具备正确的迟到处理能力迟到的元素可用side_output_late_dataOutputTag单独捕获tag id 不能为空。调试期快捷方式本地验证用execute_and_collect()同步收集结果、print()打印到标准输出、NumberSequenceSource生成测试数据跨作业复用中间结果用cache()返回的CachedDataStream并在不需要时调用invalidate()释放缓存。流批模式按数据源自动切换set_runtime_mode(RuntimeExecutionMode.AUTOMATIC)可让 Flink 依据 Source 有界/无界自动选择 BATCH/STREAMING 语义但需注意两种模式下时间与调度行为的差异见 RuntimeExecutionMode 一节。本文所有 API 清单均可在仓库内核实参考文档位于 flink-python/docs/reference/pyflink.datastream/核心实现位于 flink-python/pyflink/datastream/data_stream.pyPython 侧还可在 flink-python/pyflink/datastream/ 下找到 functions、state、window、connectors 等对应模块对应的 Java 侧实现与测试可分别在 flink-streaming-java/src/main 与 flink-python/src/test 中继续深入。赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐Flink DataStream API 编程指南从执行环境到流式应用的完整实战Flink DataStream API 编程指南从执行环境到流式应用的完整实战 Flink DataStream API 是 Apache Flink 中面大数据流处理批处理数据工程Flink DataStream API 编程指南从执行环境到流式窗口 WordCount 的完整实战Flink DataStream API 编程指南从执行环境到流式窗口 WordCount 的完整实战 DataStream API 是 Apache Fli大数据流处理批处理数据工程PyFlink Python DataStream API 入门指南从环境构建到作业提交的完整实践PyFlink Python DataStream API 入门指南从环境构建到作业提交的完整实践 本文以 Apache Flink 仓库中的官方文档 doc大数据流处理批处理数据工程上一篇Holybro Kakute H7 V2 飞控在 PX4 中的刷写、构建与配置完整指南下一篇手机也能跑Qwen3阿里开源模型实现端侧AI范式突破创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考