[GitHub] [flink] shuiqiangchen commented on a change in pull request #13803: [FLINK-19821][python] Add ProcessFunction and timer access for Python DataStream API.

[GitBox <gi...@apache.org>]

shuiqiangchen commented on a change in pull request #13803:
URL: https://github.com/apache/flink/pull/13803#discussion_r517734385



##########
File path: flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py
##########
@@ -16,43 +16,73 @@
 # limitations under the License.
 ################################################################################
 
-from pyflink.common.serialization import JsonRowSerializationSchema, \
-    JsonRowDeserializationSchema
+from pyflink.common.serialization import SimpleStringSchema
 from pyflink.common.typeinfo import Types
-from pyflink.datastream import StreamExecutionEnvironment
-from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer
+from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
+from pyflink.datastream.connectors import FlinkKafkaProducer
+from pyflink.datastream.functions import ProcessFunction, Collector
+from pyflink.table import StreamTableEnvironment
 
-from functions import m_flat_map, add_one
+from functions import MyKeySelector
 
 
 def python_data_stream_example():
     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.
+    env.set_parallelism(1)
+    env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
+    t_env = StreamTableEnvironment.create(stream_execution_environment=env)
 
-    source_type_info = Types.ROW([Types.STRING(), Types.INT()])

Review comment:
       Currently, users are not able to defined a TimestampAssigner and Watermark generator for DataStream, so, we choose to create a DataStream with row time and watermark strategy specified via a SQL DDL. It would be supported for users to defined WaterMarkStrategy in the later PR.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org