You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/10 06:51:12 UTC
[GitHub] [flink] HuangXingBo opened a new pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
HuangXingBo opened a new pull request #13369:
URL: https://github.com/apache/flink/pull/13369
## What is the purpose of the change
*This pull request will add Pandas Batch Group Aggregation Function Operator*
## Brief change log
- *Add BatchArrowPythonGroupAggregateFunctionOperator*
- *change arrow coder urn*
## Verifying this change
This change added tests and can be verified as follows:
*UT test in BatchArrowPythonGroupAggregateFunctionOperatorTest*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
----------------------------------------------------------------
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
[GitHub] [flink] dianfu closed pull request #13369: [FLINK-19173][python] Introduce BatchArrowPythonGroupAggregateFunctionOperator
Posted by GitBox <gi...@apache.org>.
dianfu closed pull request #13369:
URL: https://github.com/apache/flink/pull/13369
----------------------------------------------------------------
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
[GitHub] [flink] dianfu commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489173894
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/AbstractBatchArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.binary.BinaryRowDataUtil;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.aggregate.arrow.AbstractArrowPythonAggregateFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+/**
+ * The Abstract class of Batch Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+abstract class AbstractBatchArrowPythonAggregateFunctionOperator
Review comment:
It seems that there is no need to add an extra abstract class. Could we merge this class and BatchArrowPythonAggregateFunctionOperator into a single class?
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * The Batch Arrow Python {@link AggregateFunction} Operator for Group Aggregation.
+ */
+@Internal
+public class BatchArrowPythonGroupAggregateFunctionOperator
+ extends AbstractBatchArrowPythonAggregateFunctionOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ public BatchArrowPythonGroupAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupKey,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, pandasAggFunctions, inputType, outputType, groupKey, groupingSet, udafInputOffsets);
+ }
+
+ @Override
+ public void open() throws Exception {
+ userDefinedFunctionOutputType = new RowType(
+ outputType.getFields().subList(groupingSet.length, outputType.getFieldCount()));
+ super.open();
+ }
+
+ @Override
+ protected void invokeCurrentBatch() throws Exception {
+ if (currentBatchCount > 0) {
+ arrowSerializer.finishCurrentBatch();
+ pythonFunctionRunner.process(baos.toByteArray());
+ baos.reset();
+ checkInvokeFinishBundleByCount();
+ }
+ }
+
+ @Override
+ public void bufferInput(RowData input) throws Exception {
+ BinaryRowData currentKey = groupKeyProjection.apply(input).copy();
+ if (lastGroupKey == null) {
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ } else if (isNewKey(currentKey)) {
+ invokeCurrentBatch();
Review comment:
So it will trigger finishBundle for each key? Is it possible to trigger finishBundle for multiple keys?
##########
File path: flink-python/src/test/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperatorTest.java
##########
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.python.PythonFunctionRunner;
+import org.apache.flink.python.PythonOptions;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.runtime.operators.python.aggregate.arrow.AbstractArrowPythonAggregateFunctionOperator;
+import org.apache.flink.table.runtime.operators.python.aggregate.arrow.ArrowPythonAggregateFunctionOperatorTestBase;
+import org.apache.flink.table.runtime.utils.PassThroughPythonAggregateFunctionRunner;
+import org.apache.flink.table.runtime.utils.PythonTestUtils;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Test for {@link BatchArrowPythonGroupAggregateFunctionOperator}. These test that:
+ *
+ * <ul>
+ * <li>FinishBundle is called when checkpoint is encountered</li>
+ * <li>Watermarks are buffered and only sent to downstream when finishedBundle is triggered</li>
+ * </ul>
+ */
+public class BatchArrowPythonGroupAggregateFunctionOperatorTest
+ extends ArrowPythonAggregateFunctionOperatorTestBase {
+
+ @Test
+ public void testGroupAggregateFunction() throws Exception {
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = getTestHarness(
+ new Configuration());
+ long initialTime = 0L;
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ testHarness.open();
+
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c2", 0L), initialTime + 1));
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c4", 1L), initialTime + 2));
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c2", "c6", 2L), initialTime + 3));
+ testHarness.close();
+
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c1", 0L)));
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c2", 2L)));
+
+ assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+ }
+
+ @Test
+ public void testFinishBundleTriggeredOnCheckpoint() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setInteger(PythonOptions.MAX_BUNDLE_SIZE, 10);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = getTestHarness(conf);
+
+ long initialTime = 0L;
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ testHarness.open();
+
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c2", 0L), initialTime + 1));
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c4", 1L), initialTime + 2));
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c2", "c6", 2L), initialTime + 3));
+ // checkpoint trigger finishBundle
+ testHarness.prepareSnapshotPreBarrier(0L);
+
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c1", 0L)));
+
+ assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+
+ testHarness.close();
+
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c2", 2L)));
+
+ assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+ }
+
+ @Test
+ public void testFinishBundleTriggeredByCount() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setInteger(PythonOptions.MAX_BUNDLE_SIZE, 2);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = getTestHarness(conf);
+
+ long initialTime = 0L;
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ testHarness.open();
+
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c2", 0L), initialTime + 1));
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c1", "c2", 1L), initialTime + 2));
+ assertOutputEquals("FinishBundle should not be triggered.", expectedOutput, testHarness.getOutput());
+
+ testHarness.processElement(new StreamRecord<>(newRow(true, "c2", "c6", 2L), initialTime + 2));
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c1", 0L)));
+
+ assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+
+ testHarness.close();
+
+ expectedOutput.add(new StreamRecord<>(newRow(true, "c2", 2L)));
+
+ assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
+ }
+
+ @Test
+ public void testFinishBundleTriggeredByTime() throws Exception {
Review comment:
Should we disable "finish bundle trigger by time"?
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java
##########
@@ -38,7 +38,7 @@
private static final long serialVersionUID = 1L;
- private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:scalar_function:arrow:v1";
Review comment:
Why change this?
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.JoinedRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.arrow.serializers.ArrowSerializer;
+import org.apache.flink.table.runtime.arrow.serializers.RowDataArrowSerializer;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+public abstract class AbstractArrowPythonAggregateFunctionOperator
+ extends AbstractStatelessFunctionOperator<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:arrow:v1";
+
+ private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:pandas_aggregate_function:v1";
+
+ /**
+ * The Pandas {@link AggregateFunction}s to be executed.
+ */
+ private final PythonFunctionInfo[] pandasAggFunctions;
+
+ protected final int[] groupingSet;
+
+ protected transient ArrowSerializer<RowData> arrowSerializer;
+
+ /**
+ * The collector used to collect records.
+ */
+ protected transient StreamRecordRowDataWrappingCollector rowDataWrapper;
+
+ /**
+ * The JoinedRowData reused holding the execution result.
+ */
+ protected transient JoinedRowData reuseJoinedRow;
+
+ /**
+ * The current number of elements to be included in an arrow batch.
+ */
+ protected transient int currentBatchCount;
+
+ /**
+ * The Projection which projects the udaf input fields from the input row.
+ */
+ private transient Projection<RowData, BinaryRowData> udafInputProjection;
+
+ public AbstractArrowPythonAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, inputType, outputType, udafInputOffsets);
+ this.pandasAggFunctions = Preconditions.checkNotNull(pandasAggFunctions);
+ this.groupingSet = Preconditions.checkNotNull(groupingSet);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ rowDataWrapper = new StreamRecordRowDataWrappingCollector(output);
+ reuseJoinedRow = new JoinedRowData();
+
+ udafInputProjection = createUdafInputProjection();
+ arrowSerializer = new RowDataArrowSerializer(userDefinedFunctionInputType, userDefinedFunctionOutputType);
+ arrowSerializer.open(bais, baos);
+ currentBatchCount = 0;
+ }
+
+ @Override
+ public void dispose() throws Exception {
+ super.dispose();
+ arrowSerializer.close();
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) throws Exception {
+ RowData value = element.getValue();
+ bufferInput(value);
+ processElementInternal(value);
+ emitResults();
+ }
+
+ @Override
+ protected void checkInvokeFinishBundleByCount() throws Exception {
+ elementCount += currentBatchCount;
+ currentBatchCount = 0;
+ if (elementCount >= maxBundleSize) {
+ invokeFinishBundle();
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
Review comment:
This implementation is very similar to the implementation in the super class. It seems that we add this method just because **elementCount** was computed in **checkInvokeFinishBundleByCount**. If so, we just need to improve this, then we could avoid the duplicate code.
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.JoinedRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.arrow.serializers.ArrowSerializer;
+import org.apache.flink.table.runtime.arrow.serializers.RowDataArrowSerializer;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+public abstract class AbstractArrowPythonAggregateFunctionOperator
+ extends AbstractStatelessFunctionOperator<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:arrow:v1";
+
+ private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:pandas_aggregate_function:v1";
Review comment:
```suggestion
private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:aggregate_function:arrow:v1";
```
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.JoinedRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.arrow.serializers.ArrowSerializer;
+import org.apache.flink.table.runtime.arrow.serializers.RowDataArrowSerializer;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+public abstract class AbstractArrowPythonAggregateFunctionOperator
+ extends AbstractStatelessFunctionOperator<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:arrow:v1";
+
+ private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:pandas_aggregate_function:v1";
+
+ /**
+ * The Pandas {@link AggregateFunction}s to be executed.
+ */
+ private final PythonFunctionInfo[] pandasAggFunctions;
+
+ protected final int[] groupingSet;
+
+ protected transient ArrowSerializer<RowData> arrowSerializer;
+
+ /**
+ * The collector used to collect records.
+ */
+ protected transient StreamRecordRowDataWrappingCollector rowDataWrapper;
+
+ /**
+ * The JoinedRowData reused holding the execution result.
+ */
+ protected transient JoinedRowData reuseJoinedRow;
+
+ /**
+ * The current number of elements to be included in an arrow batch.
+ */
+ protected transient int currentBatchCount;
+
+ /**
+ * The Projection which projects the udaf input fields from the input row.
+ */
+ private transient Projection<RowData, BinaryRowData> udafInputProjection;
+
+ public AbstractArrowPythonAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, inputType, outputType, udafInputOffsets);
+ this.pandasAggFunctions = Preconditions.checkNotNull(pandasAggFunctions);
+ this.groupingSet = Preconditions.checkNotNull(groupingSet);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
Review comment:
Should we disable the timer based **finishBundle**?
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * The Batch Arrow Python {@link AggregateFunction} Operator for Group Aggregation.
+ */
+@Internal
+public class BatchArrowPythonGroupAggregateFunctionOperator
+ extends AbstractBatchArrowPythonAggregateFunctionOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ public BatchArrowPythonGroupAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupKey,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, pandasAggFunctions, inputType, outputType, groupKey, groupingSet, udafInputOffsets);
+ }
+
+ @Override
+ public void open() throws Exception {
+ userDefinedFunctionOutputType = new RowType(
+ outputType.getFields().subList(groupingSet.length, outputType.getFieldCount()));
+ super.open();
+ }
+
+ @Override
+ protected void invokeCurrentBatch() throws Exception {
+ if (currentBatchCount > 0) {
+ arrowSerializer.finishCurrentBatch();
+ pythonFunctionRunner.process(baos.toByteArray());
+ baos.reset();
+ checkInvokeFinishBundleByCount();
+ }
+ }
+
+ @Override
+ public void bufferInput(RowData input) throws Exception {
+ BinaryRowData currentKey = groupKeyProjection.apply(input).copy();
+ if (lastGroupKey == null) {
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ } else if (isNewKey(currentKey)) {
+ invokeCurrentBatch();
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ }
+ }
+
+ @Override
+ public void processElementInternal(RowData value) {
+ arrowSerializer.write(getFunctionInput(value));
+ currentBatchCount++;
+ }
+
+ @Override
+ @SuppressWarnings("ConstantConditions")
+ public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f0;
+ int length = resultTuple.f1;
+ bais.setBuffer(udafResult, 0, length);
+ int rowCount = arrowSerializer.load();
+ for (int i = 0; i < rowCount; i++) {
+ RowData input = forwardedInputQueue.poll();
Review comment:
```suggestion
RowData key = forwardedInputQueue.poll();
```
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * The Batch Arrow Python {@link AggregateFunction} Operator for Group Aggregation.
+ */
+@Internal
+public class BatchArrowPythonGroupAggregateFunctionOperator
+ extends AbstractBatchArrowPythonAggregateFunctionOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ public BatchArrowPythonGroupAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupKey,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, pandasAggFunctions, inputType, outputType, groupKey, groupingSet, udafInputOffsets);
+ }
+
+ @Override
+ public void open() throws Exception {
+ userDefinedFunctionOutputType = new RowType(
+ outputType.getFields().subList(groupingSet.length, outputType.getFieldCount()));
+ super.open();
+ }
+
+ @Override
+ protected void invokeCurrentBatch() throws Exception {
+ if (currentBatchCount > 0) {
+ arrowSerializer.finishCurrentBatch();
+ pythonFunctionRunner.process(baos.toByteArray());
+ baos.reset();
+ checkInvokeFinishBundleByCount();
+ }
+ }
+
+ @Override
+ public void bufferInput(RowData input) throws Exception {
+ BinaryRowData currentKey = groupKeyProjection.apply(input).copy();
+ if (lastGroupKey == null) {
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ } else if (isNewKey(currentKey)) {
+ invokeCurrentBatch();
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ }
+ }
+
+ @Override
+ public void processElementInternal(RowData value) {
+ arrowSerializer.write(getFunctionInput(value));
+ currentBatchCount++;
+ }
+
+ @Override
+ @SuppressWarnings("ConstantConditions")
+ public void emitResult(Tuple2<byte[], Integer> resultTuple) throws Exception {
+ byte[] udafResult = resultTuple.f0;
+ int length = resultTuple.f1;
+ bais.setBuffer(udafResult, 0, length);
+ int rowCount = arrowSerializer.load();
+ for (int i = 0; i < rowCount; i++) {
+ RowData input = forwardedInputQueue.poll();
+ reuseJoinedRow.setRowKind(input.getRowKind());
+ RowData data = arrowSerializer.read(i);
Review comment:
```suggestion
RowData result = arrowSerializer.read(i);
```
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot edited a comment on pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c07c5e36851207112dc67aad490005a3d26d39c5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489266726
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.JoinedRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.arrow.serializers.ArrowSerializer;
+import org.apache.flink.table.runtime.arrow.serializers.RowDataArrowSerializer;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+public abstract class AbstractArrowPythonAggregateFunctionOperator
+ extends AbstractStatelessFunctionOperator<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:arrow:v1";
+
+ private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:pandas_aggregate_function:v1";
+
+ /**
+ * The Pandas {@link AggregateFunction}s to be executed.
+ */
+ private final PythonFunctionInfo[] pandasAggFunctions;
+
+ protected final int[] groupingSet;
+
+ protected transient ArrowSerializer<RowData> arrowSerializer;
+
+ /**
+ * The collector used to collect records.
+ */
+ protected transient StreamRecordRowDataWrappingCollector rowDataWrapper;
+
+ /**
+ * The JoinedRowData reused holding the execution result.
+ */
+ protected transient JoinedRowData reuseJoinedRow;
+
+ /**
+ * The current number of elements to be included in an arrow batch.
+ */
+ protected transient int currentBatchCount;
+
+ /**
+ * The Projection which projects the udaf input fields from the input row.
+ */
+ private transient Projection<RowData, BinaryRowData> udafInputProjection;
+
+ public AbstractArrowPythonAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, inputType, outputType, udafInputOffsets);
+ this.pandasAggFunctions = Preconditions.checkNotNull(pandasAggFunctions);
+ this.groupingSet = Preconditions.checkNotNull(groupingSet);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ rowDataWrapper = new StreamRecordRowDataWrappingCollector(output);
+ reuseJoinedRow = new JoinedRowData();
+
+ udafInputProjection = createUdafInputProjection();
+ arrowSerializer = new RowDataArrowSerializer(userDefinedFunctionInputType, userDefinedFunctionOutputType);
+ arrowSerializer.open(bais, baos);
+ currentBatchCount = 0;
+ }
+
+ @Override
+ public void dispose() throws Exception {
+ super.dispose();
+ arrowSerializer.close();
+ }
+
+ @Override
+ public void processElement(StreamRecord<RowData> element) throws Exception {
+ RowData value = element.getValue();
+ bufferInput(value);
+ processElementInternal(value);
+ emitResults();
+ }
+
+ @Override
+ protected void checkInvokeFinishBundleByCount() throws Exception {
+ elementCount += currentBatchCount;
+ currentBatchCount = 0;
+ if (elementCount >= maxBundleSize) {
+ invokeFinishBundle();
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
Review comment:
Yes. Make sense. I will extract the logic about deciding whether there is buffered unprocessed data into a method so that we can avoid overriding `processWatermark` method
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot commented on pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c07c5e36851207112dc67aad490005a3d26d39c5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot edited a comment on pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c07c5e36851207112dc67aad490005a3d26d39c5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489264978
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/AbstractBatchArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.binary.BinaryRowDataUtil;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.aggregate.arrow.AbstractArrowPythonAggregateFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+/**
+ * The Abstract class of Batch Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+abstract class AbstractBatchArrowPythonAggregateFunctionOperator
Review comment:
I will add another two implementations of `AbstractBatchArrowPythonAggregateFunctionOperator` about Batch Group Window and Over Window. So I extract the common logic into the abstract class `AbstractBatchArrowPythonAggregateFunctionOperator`
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-693271471
@dianfu Thanks a lot for the review. I have addressed comments at the latest commit.
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489263171
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupAggregateFunctionOperator.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow.batch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * The Batch Arrow Python {@link AggregateFunction} Operator for Group Aggregation.
+ */
+@Internal
+public class BatchArrowPythonGroupAggregateFunctionOperator
+ extends AbstractBatchArrowPythonAggregateFunctionOperator {
+
+ private static final long serialVersionUID = 1L;
+
+ public BatchArrowPythonGroupAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupKey,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, pandasAggFunctions, inputType, outputType, groupKey, groupingSet, udafInputOffsets);
+ }
+
+ @Override
+ public void open() throws Exception {
+ userDefinedFunctionOutputType = new RowType(
+ outputType.getFields().subList(groupingSet.length, outputType.getFieldCount()));
+ super.open();
+ }
+
+ @Override
+ protected void invokeCurrentBatch() throws Exception {
+ if (currentBatchCount > 0) {
+ arrowSerializer.finishCurrentBatch();
+ pythonFunctionRunner.process(baos.toByteArray());
+ baos.reset();
+ checkInvokeFinishBundleByCount();
+ }
+ }
+
+ @Override
+ public void bufferInput(RowData input) throws Exception {
+ BinaryRowData currentKey = groupKeyProjection.apply(input).copy();
+ if (lastGroupKey == null) {
+ lastGroupKey = currentKey;
+ lastGroupSet = groupSetProjection.apply(input).copy();
+ forwardedInputQueue.add(lastGroupSet);
+ } else if (isNewKey(currentKey)) {
+ invokeCurrentBatch();
Review comment:
We do not call the finishBundle for each key, but pack the data of a key into an Arrow format Data.
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot edited a comment on pull request #13369: [FLINK-19173][python] Introduce BatchArrowPythonGroupAggregateFunctionOperator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
}, {
"hash" : "d3b8437a579749fea89deddce969637943d4796f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6544",
"triggerID" : "d3b8437a579749fea89deddce969637943d4796f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c07c5e36851207112dc67aad490005a3d26d39c5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424)
* d3b8437a579749fea89deddce969637943d4796f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6544)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489272675
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/arrow/ArrowPythonScalarFunctionOperator.java
##########
@@ -38,7 +38,7 @@
private static final long serialVersionUID = 1L;
- private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:scalar_function:arrow:v1";
Review comment:
The arrow coder will also be used in Pandas UDAF. It will be strange if scalar function is in coder urn. In fact, the content of the function in the coder urn is not reasonable, but for compatibility, I agree not to modify the urn of the original arrow coder, and then I will add a new urn, and also map it to the arrow coder. What do you think of this?
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot edited a comment on pull request #13369: [FLINK-19173][python] Introduce BatchArrowPythonGroupAggregateFunctionOperator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
}, {
"hash" : "d3b8437a579749fea89deddce969637943d4796f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3b8437a579749fea89deddce969637943d4796f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c07c5e36851207112dc67aad490005a3d26d39c5 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424)
* d3b8437a579749fea89deddce969637943d4796f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot commented on pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690029632
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit c07c5e36851207112dc67aad490005a3d26d39c5 (Thu Sep 10 06:53:50 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-19173).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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
[GitHub] [flink] HuangXingBo commented on a change in pull request #13369: [FLINK-19173][python] Add Pandas Batch Group Aggregation Function Operator
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #13369:
URL: https://github.com/apache/flink/pull/13369#discussion_r489268693
##########
File path: flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/AbstractArrowPythonAggregateFunctionOperator.java
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.python.aggregate.arrow;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.fnexecution.v1.FlinkFnApi;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.data.JoinedRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.python.PythonEnv;
+import org.apache.flink.table.functions.python.PythonFunctionInfo;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.ProjectionCodeGenerator;
+import org.apache.flink.table.runtime.arrow.serializers.ArrowSerializer;
+import org.apache.flink.table.runtime.arrow.serializers.RowDataArrowSerializer;
+import org.apache.flink.table.runtime.generated.GeneratedProjection;
+import org.apache.flink.table.runtime.generated.Projection;
+import org.apache.flink.table.runtime.operators.python.AbstractStatelessFunctionOperator;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * The Abstract class of Arrow Aggregate Operator for Pandas {@link AggregateFunction}.
+ */
+@Internal
+public abstract class AbstractArrowPythonAggregateFunctionOperator
+ extends AbstractStatelessFunctionOperator<RowData, RowData, RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String SCHEMA_ARROW_CODER_URN = "flink:coder:schema:arrow:v1";
+
+ private static final String PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:pandas_aggregate_function:v1";
+
+ /**
+ * The Pandas {@link AggregateFunction}s to be executed.
+ */
+ private final PythonFunctionInfo[] pandasAggFunctions;
+
+ protected final int[] groupingSet;
+
+ protected transient ArrowSerializer<RowData> arrowSerializer;
+
+ /**
+ * The collector used to collect records.
+ */
+ protected transient StreamRecordRowDataWrappingCollector rowDataWrapper;
+
+ /**
+ * The JoinedRowData reused holding the execution result.
+ */
+ protected transient JoinedRowData reuseJoinedRow;
+
+ /**
+ * The current number of elements to be included in an arrow batch.
+ */
+ protected transient int currentBatchCount;
+
+ /**
+ * The Projection which projects the udaf input fields from the input row.
+ */
+ private transient Projection<RowData, BinaryRowData> udafInputProjection;
+
+ public AbstractArrowPythonAggregateFunctionOperator(
+ Configuration config,
+ PythonFunctionInfo[] pandasAggFunctions,
+ RowType inputType,
+ RowType outputType,
+ int[] groupingSet,
+ int[] udafInputOffsets) {
+ super(config, inputType, outputType, udafInputOffsets);
+ this.pandasAggFunctions = Preconditions.checkNotNull(pandasAggFunctions);
+ this.groupingSet = Preconditions.checkNotNull(groupingSet);
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
Review comment:
I think we still need the timer based finishBundle. We can still use elementCount and time to control the frequency of sending data in terms of quantity and time
----------------------------------------------------------------
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
[GitHub] [flink] flinkbot edited a comment on pull request #13369: [FLINK-19173][python] Introduce BatchArrowPythonGroupAggregateFunctionOperator
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13369:
URL: https://github.com/apache/flink/pull/13369#issuecomment-690035137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6424",
"triggerID" : "c07c5e36851207112dc67aad490005a3d26d39c5",
"triggerType" : "PUSH"
}, {
"hash" : "d3b8437a579749fea89deddce969637943d4796f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6544",
"triggerID" : "d3b8437a579749fea89deddce969637943d4796f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3b8437a579749fea89deddce969637943d4796f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6544)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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