https://flink.apache.org/ logo
Join Slack
Powered by
I'm using PyFlink 1.15.2, with tumbling window wit...
# troubleshooting
j
I'm using PyFlink 1.15.2, with tumbling window with a custom UDAF. I received the following error:
Copy code
Pandas UDAFs are not supported in streaming mode currently.
Please check the documentation for the set of currently supported SQL features.
	at org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:77)
	at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
	at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
	at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
	at scala.collection.Iterator.foreach(Iterator.scala:937)
	at scala.collection.Iterator.foreach$(Iterator.scala:937)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
	at scala.collection.IterableLike.foreach(IterableLike.scala:70)
	at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
	at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
	at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
	at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
	at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
	at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:175)
	at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:82)
	at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:75)
	at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:306)
	at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:186)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1656)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:782)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:861)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:701)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
	at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
	at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.base/java.lang.Thread.run(Thread.java:829)
I did some googling and found a jira link [1] for version 1.12.0, the corresponding confluence page [2] suggests the following syntax:
Copy code
#  option2:  Python function
@udaf(result_type=DataTypes.FLOAT(), udaf_type="pandas")
def mean_udaf(v: pd.Series) -> float:
    return v.mean()
So, I adjusted my UDAF decorator to follow suit:
Copy code
@udaf(result_type=DataTypes.VARCHAR(6), func_type='pandas')
def trigger_source(
    xxx,
    yyy=None
):
However, when I rerun my PyFlink 1.15.2 application using the suggestions from the confluence page [2], I get a different error:
Copy code
@udaf(result_type=DataTypes.VARCHAR(6), udaf_type="pandas")
TypeError: udaf() got an unexpected keyword argument 'udaf_type'
--- [1] https://issues.apache.org/jira/browse/FLINK-19169 [2] https://cwiki.apache.org/confluence/display/FLINK/FLIP-137%3A+Support+Pandas+UDAF+in+PyFlink
d
Seems like you should try batching mode
j
I can only use PyFlink 1.13.x or 1.15.x (don't think it supports batching -- I don't see anything about batching) [1]. If there was a jira/confluence about 
pandas
during version 1.12.0, I'd like to know if I can use 
pandas
for tumbling window with custom UDAF. --- [1] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/overview/
I was able to change it to batch mode [1]:
Copy code
-env_settings = StreamExecutionEnvironment.get_execution_environment()
-table_env = StreamTableEnvironment.create(env_settings)
+env_settings = EnvironmentSettings.in_batch_mode()
+table_env = TableEnvironment.create(env_settings)
However, upon executing the same code with the above adjustment, I get an error stating the table source is not bounded:
Copy code
py4j.protocol.Py4JJavaError: An error occurred while calling o1.sqlQuery.
: org.apache.flink.table.api.ValidationException: Querying an unbounded table 'default_catalog.default_database.input_table_source' in batch mode is not allowed. The table source is unbounded.
	at org.apache.flink.table.planner.connectors.DynamicSourceUtils.validateScanSourceForBatch(DynamicSourceUtils.java:521)
	at org.apache.flink.table.planner.connectors.DynamicSourceUtils.validateScanSource(DynamicSourceUtils.java:467)
	at org.apache.flink.table.planner.connectors.DynamicSourceUtils.prepareDynamicSource(DynamicSourceUtils.java:158)
	at org.apache.flink.table.planner.connectors.DynamicSourceUtils.convertSourceToRel(DynamicSourceUtils.java:122)
	at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:118)
	at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3585)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2507)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2144)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2093)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2050)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:663)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:644)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3438)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:570)
	at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:197)
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:189)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:1240)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:1188)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertValidatedSqlNode(SqlToOperationConverter.java:345)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:238)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:105)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:675)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
	at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
	at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.base/java.lang.Thread.run(Thread.java:829)
[1] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/udfs/vectorized_python_udfs/#vectorized-scalar-functions
d
yeah I think it needs to be converted from a stream to a bounded table
I'm trying to do this myself
j
I was starting to think of that approach, but wasn't sure it was even possible. Another two ideas I had, continue using UDAF in python syntax without 
pandas
. But, I'm not sure whether I need to remove 
func_type='pandas'
or replace it? Another idea, was to maybe write the UDAF in java. But out of the three different ideas I have so far, java is my least preferred, whereas as the other/first two ideas not sure what is possible
maybe I'll just do away with 
pandas
and try to better understand how to extend the 
AggregateFunction
[1]. --- [1] https://stackoverflow.com/questions/75358446/pyflink-user-defined-aggregate-function-does-not-accumulate-values-and-merge-and
d
yeah it's a struggle to understand that correct adapters to use
j
I'm starting to use the 
AggregateFunction
[1]. Maybe something trivial by tomorrow, then hopefully better understand to expand to my original use case. Are you still actively trying to convert unbounded stream (Kafka/kinesis) to bounded? I'm not sure that is possible, without an intermediate data store, then re/reading the data, which kind of defeats the whole purpose of py/flink near/real time. But, my understanding is generally less than sub/par regarding streaming. --- [1] https://stackoverflow.com/questions/75358446/pyflink-user-defined-aggregate-function-does-not-accumulate-values-and-merge-and
d
For this particular use case I think we we will use flink in batch mode to read in S3 files instead of directly from kinesis.
3 Views
Open in Slack
PreviousNext
Powered by