HuangXingBo commented on code in PR #20375:
URL: https://github.com/apache/flink/pull/20375#discussion_r932955318
##########
flink-python/pyflink/fn_execution/embedded/converters.py:
##########
@@ -214,9 +228,23 @@ def from_field_type_proto(field_type):
[from_field_type_proto(f.type) for f in
field_type.row_schema.fields],
[f.name for f in field_type.row_schema.fields])
elif type_name == schema_type_name.BASIC_ARRAY:
- return
ListDataConverter(from_field_type_proto(field_type.collection_element_type))
+ return
ArrayDataConverter(from_field_type_proto(field_type.collection_element_type))
elif type_name == schema_type_name.MAP:
return
DictDataConverter(from_field_type_proto(field_type.map_info.key_type),
from_field_type_proto(field_type.map_info.value_type))
return IdentityDataConverter()
+
+
+def from_type_info(type_info: TypeInformation):
+ if isinstance(type_info, (PickledBytesTypeInfo, RowTypeInfo,
TupleTypeInfo)):
Review Comment:
Because pemja doesn't recognize these two types, which means that we need
data converters to transform these state data. I think we can use this as an
optimization later. WDYT
##########
flink-python/pyflink/datastream/data_stream.py:
##########
@@ -1205,40 +1211,47 @@ def __init__(self, reduce_function):
self._open_func = None
self._close_func = None
self._reduce_function = reduce_function
- self._reduce_value_state = None # type: ValueState
+ self._reduce_state = None # type: ReducingState
+ self._in_batch_execution_mode = True
+ self._has_started_key_set = set()
def open(self, runtime_context: RuntimeContext):
if self._open_func:
self._open_func(runtime_context)
- self._reduce_value_state = runtime_context.get_state(
- ValueStateDescriptor("_reduce_state" + str(uuid.uuid4()),
output_type))
- from pyflink.fn_execution.datastream.process.runtime_context
import (
- StreamingRuntimeContext)
- self._in_batch_execution_mode = \
- cast(StreamingRuntimeContext,
runtime_context)._in_batch_execution_mode
+ self._reduce_state = runtime_context.get_reducing_state(
+ ReducingStateDescriptor(
+ "_reduce_state" + str(uuid.uuid4()),
+ self._reduce_function,
+ output_type))
+
+ if python_execution_mode == "process":
+ from
pyflink.fn_execution.datastream.process.runtime_context import (
+ StreamingRuntimeContext)
+ self._in_batch_execution_mode = (
+ cast(StreamingRuntimeContext,
runtime_context)._in_batch_execution_mode)
+ else:
+ self._in_batch_execution_mode =
runtime_context.get_job_parameter(
+ "inBatchExecutionMode", "false") == "true"
def close(self):
if self._close_func:
self._close_func()
def process_element(self, value, ctx:
'KeyedProcessFunction.Context'):
- reduce_value = self._reduce_value_state.value()
- if reduce_value is not None:
- reduce_value = self._reduce_function(reduce_value, value)
- else:
- # register a timer for emitting the result at the end when
this is the
- # first input for this key
- if self._in_batch_execution_mode:
+ self._reduce_state.add(value)
+ if self._in_batch_execution_mode:
+ key = ctx.get_current_key()
+ if isinstance(key, list):
+ key = tuple(key)
+ if key not in self._has_started_key_set:
ctx.timer_service().register_event_time_timer(0x7fffffffffffffff)
- reduce_value = value
- self._reduce_value_state.update(reduce_value)
- if not self._in_batch_execution_mode:
- # only emitting the result when all the data for a key is
received
- yield reduce_value
+ self._has_started_key_set.add(key)
Review Comment:
Yes. Good catch
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
