also I've other issue with same code. When I'm am using tumble to group by 1 min. it throws any error. result = table\ .window(Tumble.over(lit(1).minutes).on(col('rowtime')).alias("w"))\ .group_by(col("w"),col("c"))\ .select(col("w"),col("c"), col("b").count.alias('d')) result.execute().print()

: org.apache.flink.table.api.ValidationException: A group window expects a time attribute for grouping in a stream environment. at org.apache.flink.table.operations.utils.AggregateOperationFactory.validateStreamTimeAttribute(AggregateOperationFactory.java:327) at org.apache.flink.table.operations.utils.AggregateOperationFactory.validateTimeAttributeType(AggregateOperationFactory.java:307) at org.apache.flink.table.operations.utils.AggregateOperationFactory.getValidatedTimeAttribute(AggregateOperationFactory.java:300) at org.apache.flink.table.operations.utils.AggregateOperationFactory.createResolvedWindow(AggregateOperationFactory.java:265) at org.apache.flink.table.operations.utils.OperationTreeBuilder.windowAggregate(OperationTreeBuilder.java:262) at org.apache.flink.table.api.internal.TableImpl$WindowGroupedTableImpl.select(TableImpl.java:640) 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:834)

You could also run it in batch mode and then you will only see the final result.

Re the exception: You are running the job in streaming mode and row time column(not just Timestamp type) must be part of the grouping key. Could refer to https://github.com/apache/flink/blob/master/flink-python/pyflink/examples/table/windowing/tumble_window.py as an example.

However, coming back to Tumble Window example. I'm using row time column (which is timestamp) as part of grouping key if you see my example.

Why I'm using this way because, let me give background of what I'm trying to achieve.

In my current project, we are trying to connect kafka using table api kafka connector and sink with another kafka topic.

for example using this https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/kafka/

we want to achieve this using table environment created using streaming mode

for example the code looks like this

from pyflink.table import EnvironmentSettings, TableEnvironment # create a stream TableEnvironment env_settings = EnvironmentSettings.in_streaming_mode() table_env = TableEnvironment.create(env_settings) table_env.execute_sql(""" CREATE TABLE random_source (

@timestamp

BIGINT, ts as TO_TIMESTAMP(FROM_UNIXTIME(

@timestamp

)), id BIGINT, data TINYINT ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior_source', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'testGroup', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json' ) """) table = table_env.from_path("random_source") table_env.execute_sql(""" CREATE TABLE sink_table ( one_minute_window timestamp, id BIGINT, data_count TINYINT ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'user_behavior_sink', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'testGroup', 'format' = 'json' ) """) sink_table = table_env.from_path("sink_table") #table.execute().print() result = table\ .window(Tumble.over(lit(1).minutes).on(col('ts')).alias("one_minute_window"))\ .group_by(col("one_minute_window"),col("id"))\ .select(col("one_minute_window"),col("id"), col("data").count.alias('data_count')) # develop code to sink with another kafka topic result.execute().print() # for now print the results and see. #result1 = result.execute_insert("sink_table")

The code works perfectly. But only the problem it sinks old records also op=-U

The code works perfectly. But only the problem it sinks old records also op=-U

Yes, this is by design as -U message is also part of the changelog. Maybe you could try upsert-kafka: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/upsert-kafka/

I'm new to Apache Flink and trying to understand and implementing the solutions on fly. It looks like I'm missing lot of basic stuff. But your information what you provided me is awesome. Let me go through it and understand.

Hi @Dian Fu I am very close. But not sure where I am making mistake. I am attaching sample code what I am using and trying to develop.

test.py

when I run I am getting this error.

: org.apache.flink.table.api.ValidationException: Cannot resolve field [one_minute_window], input field list:[type, EXPR$0].

I want to ask one more question which is not relevant to above thread issue. However if you can help me, that will be Awesome! Thanks and much appreciated.

test_case_1.py

In the above example, I've prepared one sample data and created table api.

for that table I want to write a group by logic using table api and generate the output as show in below format. Not sure what is the best idea to develop the code.

I want to group the data by "company" and remaining are measures.

Much appreciated if you can help me on this. Thanks a lot.

For your requirements, I have written a simple example:

import json

import pandas as pd
from pyflink.common import Configuration

from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes

# create a batch TableEnvironment
from pyflink.table.expressions import col
from pyflink.table.udf import udaf

conf = Configuration()
env_settings = EnvironmentSettings.in_batch_mode()
table_env = TableEnvironment.create(env_settings)

table = table_env.from_elements(
    [(1, 'apple', 200, "a"),
     (2, 'samsung', 300, "s"),
     (3, 'nokia', 400, "n"),
     (1, 'apple', 300, "a1"),
     (2, 'samsung', 400, "s1"),
     (3, 'nokia', 500, "n1"),
     (1, 'apple', 100, "a2")
    ],
    ['id', 'company', 'price', 'category'])


@udaf(result_type=DataTypes.STRING(), func_type="pandas")
def metadata(price: pd.Series, category: pd.Series):
    result = {
        'count': price.size,
        'price': price.tolist(),
        'category': category.tolist()
    }
    return str(json.dumps(result))


@udaf(result_type=DataTypes.BIGINT(), func_type="pandas")
def total(price: pd.Series):
    return int(price.sum())


table = table.group_by(col('id'), col('company')) \
    .select(col('company'),
            total(col('price')).alias('total'),
            col('id'),
            metadata(col('price'), col('category')).alias('metadata'))

table.execute().print()

# with table.execute().collect() as results:
#     for result in results:
#         print(result)

It prints the above results. For the last column, the results are truncated. There is a JIRA to improve this which allows to configure the max display column width: https://issues.apache.org/jira/browse/FLINK-30025

Nice idea on the second question. I'll sure try it.

after adding col('five_seconds_window').start,col('five_seconds_window').end it doesn't throw error

so far good.

But it didn't prints the results when I tried to print it or sink to another topic

for example

I defined the kafka topic properties like this

In above screenshot if you see there are not results it is producing

However when I stop doing window group by and print the results of table directly. It shows the results of incoming records of kafka topic

Here is another screenshot of it

I'm very very close to achieve this solution all because of you, once I've this group results then my idea is to store the results in the format what you suggested to me for my second question ( metadata column with json format). I am not able to run the sample code in my local machine (windows) using python virtual environment set up. However I've a docker container with flink and python and I tried it and worked like a charm. Thank you. But when I changed the env setting to stream mode to the sample example what you provided me, like below code then it throws error. I'm enclosing the error too. Because down the line I want to use the json format conversion on incoming messages of kafka topic through window aggregation. env_settings = EnvironmentSettings.in_streaming_mode() table_env = TableEnvironment.create(env_settings) ------------------------- error ---- pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query: FlinkLogicalSink(table=[anonymous_collect$2], fields=[company, total, id, metadata]) +- FlinkLogicalCalc(select=[company, EXPR$0 AS total, id, EXPR$1 AS metadata]) +- FlinkLogicalAggregate(group=[{0, 1}], EXPR$0=[*org.apache.flink.table.functions.python.PythonAggregateFunction$2595d75109319bc0ebcb9b7a185a1be8*($2)], EXPR$1=[*org.apache.flink.table.functions.python.PythonAggregateFunction$c0b392a2af72b7d0801546c286cbee53*($2, $3)]) +- FlinkLogicalTableSourceScan(table=[[anonymous_python-input-format$1]], fields=[id, company, price, category]) Pandas UDAFs are not supported in streaming mode currently. Please check the documentation for the set of currently supported SQL features.

I'm also enclosing the error which I got when I'm running locally on windows through python. Toovercome the problem I reffered this solution and didn't work when I run set no_proxy=localhost https://stackoverflow.com/questions/68015759/whats-wrong-with-my-pyflink-setup-that-python-udfs-throw-py4j-exceptions

error_for_dev_json_format.txt

So Pandas can be used for GroupBy Window Aggregation.

Awesome!

currently I've results like this

It throws an error with no message

py4j.protocol.Py4JJavaError: An error occurred while calling o146.execute. : java.lang.NullPointerException at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:59) at org.apache.flink.table.functions.python.PythonFunctionInfo.<init>(PythonFunctionInfo.java:45) at org.apache.flink.table.functions.python.PythonAggregateFunctionInfo.<init>(PythonAggregateFunctionInfo.java:36) at org.apache.flink.table.planner.plan.nodes.exec.utils.CommonPythonUtil.extractPythonAggregateFunctionInfosFromAggregateCall(CommonPythonUtil.java:262) at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecPythonGroupWindowAggregate.createPandasPythonStreamWindowGroupOneInputTransformation(StreamExecPythonGroupWindowAggregate.java:406) at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecPythonGroupWindowAggregate.translateToPlanInternal(StreamExecPythonGroupWindowAggregate.java:297) at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:158) at org.apache.flink.table.planner.plan.nodes.exec.ExecEdge.translateToPlan(ExecEdge.java:257) at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecCalc.translateToPlanInternal(CommonExecCalc.java:94) at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:158) at org.apache.flink.table.planner.plan.nodes.exec.ExecEdge.translateToPlan(ExecEdge.java:257) at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:145) at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:158) at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:85) at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) 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.TraversableLike.map(TraversableLike.scala:233) at scala.collection.TraversableLike.map$(TraversableLike.scala:226) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:84) at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:197) at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1733) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:874) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1382) at org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:475) 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:834)

Finally I understood how the Apache Flink streaming and batch processing is working. It made me to read and understand lot of articles.

But at the same time when there is any error it is getting more default to trace it out. Because few errors are not well described or I am not up to level to understand at this point.

@Dian Fu last and final question to close of this thread. Metadata column doesn't show up fully. It is truncating the data due to size. Is there any way to show it fully while printing in the console.

Thank you very very much for teaching me, much appreciated for you patience. Have a wonderful day.

Since you are running a streaming job, I guess finally you want to output the data to an external storage. If that’s the case, it should not be a big problem here.