Hi, Sumeet
Thanks you for the sharing. As Dian suggested, I think you could use b as
your `group_by`'s key and so the b could be output directly.
I think it is more simple.
Best,
Guowei
On Mon, Apr 19, 2021 at 7:31 PM Dian Fu <dian0511...@gmail.com> wrote:
> Hi Sumeet,
>
> Thanks for the sharing.
>
> Then I guess you could use `.group_by(col('w'), input.a, input.b)`. Since
> the value for input.a is always the same, it’s equal to group_by(col(‘w')
> , input.b) logically. The benefit is that you could access input.a
> directly in the select clause.
>
> Regards,
> Dian
>
> 2021年4月19日 下午6:29，Sumeet Malhotra <sumeet.malho...@gmail.com> 写道：
>
> Hi Guowei,
>
> Let me elaborate the use case with an example.
>
> Sample input table looks like this:
>
> time a b c
> -----------------
> t0 a0 b0 1
> t1 a0 b1 2
> t2 a0 b2 3
> t3 a0 b0 6
> t4 a0 b1 7
> t5 a0 b2 8
>
> Basically, every time interval there are new readings from a fixed set of
> sensors (b0, b1 and b2). All these rows have a few constant fields
> representing metadata about the input (a0).
>
> Desired output for every time interval is the average reading for every
> sensor (b0, b1, b2), along with the constant metadata (a0):
>
> a0 b0 avg(c)
> a0 b1 avg(c)
> a0 b2 avg(c)
>
> This is what I was trying to build using a simple Tumble window:
>
> input.window(Tumble.over(lit(10).seconds).on(input.Timestamp).alias("w"))
> \
> .group_by(col('w'), input.b) \
> .select(
> input.a, <=== constant metadata field,
> same for every input record
> input.b, <=== group_by field, to
> compute averages
> input.c.avg.alias('avg_value')) \
> .execute_insert('MySink') \
> .wait()
>
> The example above is highly simplified, but I hope it explains what I'm
> trying to achieve.
>
> Thanks,
> Sumeet
>
>
> On Mon, Apr 19, 2021 at 3:21 PM Dian Fu <dian0511...@gmail.com> wrote:
>
>> Hi Sumeet,
>>
>> 1) Regarding to the above exception, it’s a known issue and has been
>> fixed in FLINK-21922 <https://issues.apache.org/jira/browse/FLINK-21922>
>> [1]. It
>> will be available in the coming 1.12.3. You could also cherry-pick that fix
>> to 1.12.2 and build from source following the instruction described in [2]
>> if needed.
>>
>> 2) Regarding to your requirements, could you describe what you want to do
>> with group window or over window?
>> For group window(e.g. tumble window, hop window, session window, etc), it
>> will output one row for multiple inputs belonging to the same window. You
>> could not just passing through it from input to sink as it is
>> non-determinitic which row to use as there are multiple input rows. That’s
>> the reason why you have to declare a field in the group by clause if you
>> want to access it directly in the select clause. For over window, it will
>> output one row for each input and so you could pass through it directly.
>>
>> [1] https://issues.apache.org/jira/browse/FLINK-21922.
>> [2]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/flinkDev/building.html#build-pyflink
>>
>>
>> 2021年4月19日 下午5:16，Sumeet Malhotra <sumeet.malho...@gmail.com> 写道：
>>
>> Thanks Guowei. I'm trying out Over Windows, as follows:
>>
>> input \
>> .over_window(
>> Over.partition_by(col(input.a)) \
>> .order_by(input.Timestamp) \
>> .preceding(lit(10).seconds) \
>> .alias('w')) \
>> .select(
>> input.b,
>> input.c.avg.over(col('w'))) \
>> .execute_insert('MySink') \
>> .wait()
>>
>> But running into following exception:
>>
>> py4j.protocol.Py4JError: An error occurred while calling
>> z:org.apache.flink.table.api.Over.partitionBy. Trace:
>> org.apache.flink.api.python.shaded.py4j.Py4JException: Method
>> partitionBy([class org.apache.flink.table.api.ApiExpression]) does not exist
>>
>> Is there any extra Jar that needs to be included for Over Windows. From
>> the code it doesn't appear so.
>>
>> Thanks,
>> Sumeet
>>
>>
>> On Mon, Apr 19, 2021 at 1:10 PM Guowei Ma <guowei....@gmail.com> wrote:
>>
>>> Hi, Sumeet
>>>
>>> Maybe you could try the Over Windows[1], which could keep the
>>> "non-group-key" column.
>>>
>>> [1]
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#over-windows
>>>
>>> Best,
>>> Guowei
>>>
>>>
>>> On Mon, Apr 19, 2021 at 3:25 PM Sumeet Malhotra <
>>> sumeet.malho...@gmail.com> wrote:
>>>
>>>> Thanks Guowei! Regarding "input.c.avg" you're right, that doesn't cause
>>>> any issues. It's only when I want to use "input.b".
>>>>
>>>> My use case is to basically emit "input.b" in the final sink as is, and
>>>> not really perform any aggregation on that column - more like pass through
>>>> from input to sink. What's the best way to achieve this? I was thinking
>>>> that making it part of the select() clause would do it, but as you said
>>>> there needs to be some aggregation performed on it.
>>>>
>>>> Thanks,
>>>> Sumeet
>>>>
>>>>
>>>> On Mon, Apr 19, 2021 at 12:26 PM Guowei Ma <guowei....@gmail.com>
>>>> wrote:
>>>>
>>>>> Hi, Sumeet
>>>>> For "input.b" I think you should aggregate the non-group-key
>>>>> column[1].
>>>>> But I am not sure why the "input.c.avg.alias('avg_value')" has
>>>>> resolved errors. Would you mind giving more detailed error information?
>>>>>
>>>>> [1]
>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#group-windows
>>>>>
>>>>> Best,
>>>>> Guowei
>>>>>
>>>>>
>>>>> On Mon, Apr 19, 2021 at 2:24 PM Sumeet Malhotra <
>>>>> sumeet.malho...@gmail.com> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I have a use case where I'm creating a Tumbling window as follows:
>>>>>>
>>>>>> "input" table has columns [Timestamp, a, b, c]
>>>>>>
>>>>>> input \
>>>>>>
>>>>>> .window(Tumble.over(lit(10).seconds).on(input.Timestamp).alias("w")) \
>>>>>> .group_by(col('w'), input.a) \
>>>>>> .select(
>>>>>> col('w').start.alias('window_start'),
>>>>>> col('w').end.alias('window_end'),
>>>>>> input.b,
>>>>>> input.c.avg.alias('avg_value')) \
>>>>>> .execute_insert('MySink') \
>>>>>> .wait()
>>>>>>
>>>>>> This throws an exception that it cannot resolve the fields "b" and
>>>>>> "c" inside the select statement. If I mention these column names inside
>>>>>> the
>>>>>> group_by() statement as follows:
>>>>>>
>>>>>> .group_by(col('w'), input.a, input.b, input.c)
>>>>>>
>>>>>> then the column names in the subsequent select statement can be
>>>>>> resolved.
>>>>>>
>>>>>> Basically, unless the column name is explicitly made part of the
>>>>>> group_by() clause, the subsequent select() clause doesn't resolve it.
>>>>>> This
>>>>>> is very similar to the example from Flink's documentation here [1]:
>>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#overview--examples,
>>>>>> where a similar procedure works.
>>>>>>
>>>>>> Any idea how I can access columns from the input stream, without
>>>>>> having to mention them in the group_by() clause? I really don't want to
>>>>>> group the results by those fields, but they need to be written to the
>>>>>> sink
>>>>>> eventually.
>>>>>>
>>>>>> Thanks,
>>>>>> Sumeet
>>>>>>
>>>>>
>>
>
