This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8141d7d8812901688e376553ed7bee0b2e1b8630 Author: Martijn Visser <[email protected]> AuthorDate: Fri Jul 3 16:24:20 2026 +0200 [FLINK-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink Rewrite the job on the bounded FileSource and unified FileSink and re-enable the case. Generated-by: Claude Code (Fable 5) --- .../python/datastream/data_stream_job.py | 57 ++++--- .../test-scripts/test_pyflink.sh | 188 ++++++--------------- 2 files changed, 86 insertions(+), 159 deletions(-) diff --git a/flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py b/flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py index 5c85a5756af..6c80a4ec66b 100644 --- a/flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py +++ b/flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py @@ -16,45 +16,60 @@ # limitations under the License. ################################################################################ +import json +import sys from typing import Any -from pyflink.common import Duration -from pyflink.common.serialization import SimpleStringSchema +from pyflink.common import Duration, Encoder, Row from pyflink.common.typeinfo import Types from pyflink.common.watermark_strategy import TimestampAssigner, WatermarkStrategy from pyflink.datastream import StreamExecutionEnvironment -from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer -from pyflink.datastream.formats.json import JsonRowDeserializationSchema +from pyflink.datastream.connectors.file_system import (FileSink, FileSource, RollingPolicy, + StreamFormat) from pyflink.datastream.functions import KeyedProcessFunction from functions import MyKeySelector -def python_data_stream_example(): +def python_data_stream_example(input_path: str, output_path: str): env = StreamExecutionEnvironment.get_execution_environment() - # Set the parallelism to be one to make sure that all data including fired timer and normal data - # are processed by the same worker and the collected result would be in order which is good for - # assertion. + # Process everything on one worker so the timer behavior is deterministic and the output + # lands in a single part file. env.set_parallelism(1) + # No periodic watermarks: current_watermark() stays at Long.MIN_VALUE until the bounded + # source emits MAX_WATERMARK at end of input, making the fired-timer output deterministic. + env.get_config().set_auto_watermark_interval(0) type_info = Types.ROW_NAMED(['createTime', 'orderId', 'payAmount', 'payPlatform', 'provinceId'], [Types.LONG(), Types.LONG(), Types.DOUBLE(), Types.INT(), Types.INT()]) - json_row_schema = JsonRowDeserializationSchema.builder().type_info(type_info).build() - kafka_props = {'bootstrap.servers': 'localhost:9092', 'group.id': 'pyflink-e2e-source'} - kafka_consumer = FlinkKafkaConsumer("timer-stream-source", json_row_schema, kafka_props) - kafka_producer = FlinkKafkaProducer("timer-stream-sink", SimpleStringSchema(), kafka_props) + source = FileSource.for_record_stream_format(StreamFormat.text_line_format(), input_path) \ + .process_static_file_set().build() - watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5))\ - .with_timestamp_assigner(KafkaRowTimestampAssigner()) + watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)) \ + .with_timestamp_assigner(PaymentTimestampAssigner()) - kafka_consumer.set_start_from_earliest() - ds = env.add_source(kafka_consumer).assign_timestamps_and_watermarks(watermark_strategy) - ds.key_by(MyKeySelector(), key_type=Types.LONG()) \ + sink = FileSink.for_row_format(output_path, Encoder.simple_string_encoder()) \ + .with_rolling_policy(RollingPolicy.default_rolling_policy()).build() + + # Watermarks are assigned after parsing because the timestamp field is not available on the + # raw text lines emitted by the source. + ds = env.from_source(source, WatermarkStrategy.no_watermarks(), 'payment_msg_source') + ds.map(parse_payment_msg, output_type=type_info) \ + .assign_timestamps_and_watermarks(watermark_strategy) \ + .key_by(MyKeySelector(), key_type=Types.LONG()) \ .process(MyProcessFunction(), output_type=Types.STRING()) \ - .add_sink(kafka_producer) - env.execute_async("test data stream timer") + .sink_to(sink) + # The committer commits all pending files at end of input when checkpointing is disabled, + # so no checkpoint config is needed for the output to materialize. + env.execute('test data stream timer') + + +def parse_payment_msg(line: str) -> Row: + msg = json.loads(line) + return Row(msg['createTime'], msg['orderId'], msg['payAmount'], msg['payPlatform'], + msg['provinceId']) class MyProcessFunction(KeyedProcessFunction): @@ -70,11 +85,11 @@ class MyProcessFunction(KeyedProcessFunction): yield "On timer timestamp: " + str(timestamp) -class KafkaRowTimestampAssigner(TimestampAssigner): +class PaymentTimestampAssigner(TimestampAssigner): def extract_timestamp(self, value: Any, record_timestamp: int) -> int: return int(value[0]) if __name__ == '__main__': - python_data_stream_example() + python_data_stream_example(sys.argv[1], sys.argv[2]) diff --git a/flink-end-to-end-tests/test-scripts/test_pyflink.sh b/flink-end-to-end-tests/test-scripts/test_pyflink.sh index e12dd5566e0..6fbbeb45daa 100755 --- a/flink-end-to-end-tests/test-scripts/test_pyflink.sh +++ b/flink-end-to-end-tests/test-scripts/test_pyflink.sh @@ -19,56 +19,11 @@ set -Eeuo pipefail -KAFKA_VERSION="3.2.3" -CONFLUENT_VERSION="7.5.3" -CONFLUENT_MAJOR_VERSION="7.5" -# Check the Confluent Platform <> Apache Kafka compatibility matrix when updating KAFKA_VERSION -KAFKA_SQL_VERSION="universal" -SQL_JARS_DIR=${END_TO_END_DIR}/flink-sql-client-test/target/sql-jars -KAFKA_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "kafka" ) - -function create_data_stream_kafka_source { - topicName="test-python-data-stream-source" - create_kafka_topic 1 1 $topicName - - echo "Sending messages to Kafka..." - - send_messages_to_kafka '{"f0": "a", "f1": 1}' $topicName - send_messages_to_kafka '{"f0": "ab", "f1": 2}' $topicName - send_messages_to_kafka '{"f0": "abc", "f1": 3}' $topicName - send_messages_to_kafka '{"f0": "abcd", "f1": 4}' $topicName - send_messages_to_kafka '{"f0": "abcde", "f1": 5}' $topicName -} - -function sort_msg { - arr=() - while read line - do - value=$line - arr+=("$value") - done <<< "$1" - IFS=$'\n' sorted=($(sort <<< "${arr[*]}")); unset IFS - echo "${sorted[*]}" -} - CURRENT_DIR=`cd "$(dirname "$0")" && pwd -P` source "${CURRENT_DIR}"/common.sh -source "${CURRENT_DIR}"/kafka_sql_common.sh \ - ${KAFKA_VERSION} \ - ${CONFLUENT_VERSION} \ - ${CONFLUENT_MAJOR_VERSION} \ - ${KAFKA_SQL_VERSION} function test_clean_up { stop_cluster - # Note: The 'data_stream_job.py' still uses the Kafka legacy source, - # so we temporarily disable this case because we plan to address this - # in future updates. Once we align with the upcoming PyFlink and Flink 2.0 - # compatibility efforts, we will remove this temporary disable. - # - # So we should also skip stopping the Kafka cluster because we did not start the Kafka cluster. - - # stop_kafka_cluster } on_exit test_clean_up @@ -206,100 +161,57 @@ if [[ ${debug_line_cnt} -lt 0 ]]; then EXIT_CODE=1 fi -# Note: The 'data_stream_job.py' still uses the Kafka legacy source, -# so we temporarily disable this case because we plan to address this -# in future updates. Once we align with the upcoming PyFlink and Flink 2.0 -# compatibility efforts, we will remove this temporary disable. +echo "Test PyFlink DataStream job:" + +DATA_STREAM_INPUT_DIR="${TEST_DATA_DIR}/timer-stream-input" +DATA_STREAM_OUTPUT_DIR="${TEST_DATA_DIR}/timer-stream-output" +mkdir -p "${DATA_STREAM_INPUT_DIR}" "${DATA_STREAM_OUTPUT_DIR}" + +cat > "${DATA_STREAM_INPUT_DIR}/payment_msgs.jsonl" << 'EOF' +{"createTime": 1603679413000, "orderId": 1603679414, "payAmount": 83685.44904332698, "payPlatform": 0, "provinceId": 3} +{"createTime": 1603679426000, "orderId": 1603679427, "payAmount": 30092.50657757042, "payPlatform": 0, "provinceId": 1} +{"createTime": 1603679427000, "orderId": 1603679428, "payAmount": 62644.01719293056, "payPlatform": 0, "provinceId": 6} +{"createTime": 1603679428000, "orderId": 1603679429, "payAmount": 6449.806795118451, "payPlatform": 0, "provinceId": 2} +{"createTime": 1603679491000, "orderId": 1603679492, "payAmount": 41108.36128417494, "payPlatform": 0, "provinceId": 0} +{"createTime": 1603679492000, "orderId": 1603679493, "payAmount": 64882.44233197067, "payPlatform": 0, "provinceId": 4} +{"createTime": 1603679521000, "orderId": 1603679522, "payAmount": 81648.80712644062, "payPlatform": 0, "provinceId": 3} +{"createTime": 1603679522000, "orderId": 1603679523, "payAmount": 81861.73063103345, "payPlatform": 0, "provinceId": 4} +EOF -# echo "Test PyFlink DataStream job:" -# -# # Prepare Kafka for the DataStream job -# echo "Preparing Kafka..." -# -# # Setup Kafka distribution -# setup_kafka_dist -# -# # Start Kafka cluster -# start_kafka_cluster -# -# # End to end test for DataStream ProcessFunction with timer -# # Creating necessary Kafka topics for the test -# create_kafka_topic 1 1 timer-stream-source -# create_kafka_topic 1 1 timer-stream-sink -# -# # Preparing a set of payment messages in JSON format for the test -# PAYMENT_MSGS='{"createTime": 1603679413000, "orderId": 1603679414, "payAmount": 83685.44904332698, "payPlatform": 0, "provinceId": 3} -# {"createTime": 1603679426000, "orderId": 1603679427, "payAmount": 30092.50657757042, "payPlatform": 0, "provinceId": 1} -# {"createTime": 1603679427000, "orderId": 1603679428, "payAmount": 62644.01719293056, "payPlatform": 0, "provinceId": 6} -# {"createTime": 1603679428000, "orderId": 1603679429, "payAmount": 6449.806795118451, "payPlatform": 0, "provinceId": 2} -# {"createTime": 1603679491000, "orderId": 1603679492, "payAmount": 41108.36128417494, "payPlatform": 0, "provinceId": 0} -# {"createTime": 1603679492000, "orderId": 1603679493, "payAmount": 64882.44233197067, "payPlatform": 0, "provinceId": 4} -# {"createTime": 1603679521000, "orderId": 1603679522, "payAmount": 81648.80712644062, "payPlatform": 0, "provinceId": 3} -# {"createTime": 1603679522000, "orderId": 1603679523, "payAmount": 81861.73063103345, "payPlatform": 0, "provinceId": 4}' - -# function send_msg_to_kafka { -# while read line -# do -# send_messages_to_kafka "$line" "timer-stream-source" -# sleep 1 -# done <<< "$1" -# } - -# function read_msg_from_kafka { -# $KAFKA_DIR/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning \ -# --max-messages $1 \ -# --topic $2 \ -# --consumer-property group.id=$3 --timeout-ms 90000 2> /dev/null -# } - -# send_msg_to_kafka "${PAYMENT_MSGS[*]}" - -# JOB_ID=$(${FLINK_DIR}/bin/flink run \ -# -pyfs "${FLINK_PYTHON_TEST_DIR}/python/datastream" \ -# -pyreq "${REQUIREMENTS_PATH}" \ -# -pyarch "${TEST_DATA_DIR}/venv.zip" \ -# -pyexec "venv.zip/.uv/bin/python" \ -# -pyclientexec "venv.zip/.uv/bin/python" \ -# -pym "data_stream_job" \ -# -j "${KAFKA_SQL_JAR}") - -# echo "${JOB_ID}" -# JOB_ID=`echo "${JOB_ID}" | sed 's/.* //g'` - -# wait_job_running ${JOB_ID} - -# echo "Reading kafka messages..." -# READ_MSG=$(read_msg_from_kafka 16 timer-stream-sink pyflink-e2e-test-timer) - -# We use env.execute_async() to submit the job, cancel it after fetched results. -# cancel_job "${JOB_ID}" - -# EXPECTED_MSG='Current key: 1603679414, orderId: 1603679414, payAmount: 83685.44904332698, timestamp: 1603679413000 -# Current key: 1603679427, orderId: 1603679427, payAmount: 30092.50657757042, timestamp: 1603679426000 -# Current key: 1603679428, orderId: 1603679428, payAmount: 62644.01719293056, timestamp: 1603679427000 -# Current key: 1603679429, orderId: 1603679429, payAmount: 6449.806795118451, timestamp: 1603679428000 -# Current key: 1603679492, orderId: 1603679492, payAmount: 41108.36128417494, timestamp: 1603679491000 -# Current key: 1603679493, orderId: 1603679493, payAmount: 64882.44233197067, timestamp: 1603679492000 -# Current key: 1603679522, orderId: 1603679522, payAmount: 81648.80712644062, timestamp: 1603679521000 -# Current key: 1603679523, orderId: 1603679523, payAmount: 81861.73063103345, timestamp: 1603679522000 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308 -# On timer timestamp: -9223372036854774308' - -# EXPECTED_MSG=$(sort_msg "${EXPECTED_MSG[*]}") -# SORTED_READ_MSG=$(sort_msg "${READ_MSG[*]}") - -# if [[ "${EXPECTED_MSG[*]}" != "${SORTED_READ_MSG[*]}" ]]; then -# echo "Output from Flink program does not match expected output." -# echo -e "EXPECTED Output: --${EXPECTED_MSG[*]}--" -# echo -e "ACTUAL: --${SORTED_READ_MSG[*]}--" -# exit 1 -# fi +PYFLINK_CLIENT_EXECUTABLE=${PYTHON_EXEC} "${FLINK_DIR}/bin/flink" run \ + -pyfs "${FLINK_PYTHON_TEST_DIR}/python/datastream" \ + -pyreq "${REQUIREMENTS_PATH}" \ + -pyarch "${TEST_DATA_DIR}/venv.zip" \ + -pyexec "venv.zip/.uv/bin/python" \ + -pym "data_stream_job" \ + "${DATA_STREAM_INPUT_DIR}/payment_msgs.jsonl" "${DATA_STREAM_OUTPUT_DIR}" + +sort > "${TEST_DATA_DIR}/expected_data_stream_result" << 'EOF' +Current key: 1603679414, orderId: 1603679414, payAmount: 83685.44904332698, timestamp: 1603679413000 +Current key: 1603679427, orderId: 1603679427, payAmount: 30092.50657757042, timestamp: 1603679426000 +Current key: 1603679428, orderId: 1603679428, payAmount: 62644.01719293056, timestamp: 1603679427000 +Current key: 1603679429, orderId: 1603679429, payAmount: 6449.806795118451, timestamp: 1603679428000 +Current key: 1603679492, orderId: 1603679492, payAmount: 41108.36128417494, timestamp: 1603679491000 +Current key: 1603679493, orderId: 1603679493, payAmount: 64882.44233197067, timestamp: 1603679492000 +Current key: 1603679522, orderId: 1603679522, payAmount: 81648.80712644062, timestamp: 1603679521000 +Current key: 1603679523, orderId: 1603679523, payAmount: 81861.73063103345, timestamp: 1603679522000 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +On timer timestamp: -9223372036854774308 +EOF + +find "${DATA_STREAM_OUTPUT_DIR}" -type f -name "part-*" -exec cat {} + | sort \ + > "${TEST_DATA_DIR}/actual_data_stream_result" + +if ! diff "${TEST_DATA_DIR}/expected_data_stream_result" "${TEST_DATA_DIR}/actual_data_stream_result"; then + echo "Output from PyFlink DataStream job does not match expected output." + exit 1 +fi # clean up python env "${FLINK_PYTHON_DIR}/dev/lint-python.sh" -r
