Support Python Scalar Function

dianfu · Aug 12, 2019 · 93f41ba · 93f41ba
1 parent 0cda582
commit 93f41ba
Show file tree

Hide file tree

Showing 54 changed files with 4,172 additions and 15 deletions.
diff --git a/.gitignore b/.gitignore
@@ -35,6 +35,8 @@ flink-python/dev/download
 flink-python/dev/.conda/
 flink-python/dev/log/
 flink-python/dev/.stage.txt
+flink-python/pyflink/.eggs/
+flink-python/pyflink/fn_execution/*_pb2.py
 atlassian-ide-plugin.xml
 out/
 /docs/api

diff --git a/flink-core/src/main/java/org/apache/flink/types/Row.java b/flink-core/src/main/java/org/apache/flink/types/Row.java
@@ -167,4 +167,28 @@ public static Row project(Row row, int[] fields) {
 		}
 		return newRow;
 	}
+
+	public static Row join(Row first, Row... remainings) {
+		int newLength = first.fields.length;
+		for (Row remaining : remainings) {
+			newLength += remaining.fields.length;
+		}
+
+		final Row joinedRow = new Row(newLength);
+		int index = 0;
+
+		// copy the first row
+		for (int i = 0; i < first.fields.length; i++) {
+			joinedRow.fields[index++] = first.fields[i];
+		}
+
+		// copy the remaining rows
+		for (Row remaining : remainings) {
+			for (int i = 0; i < remaining.fields.length; i++) {
+				joinedRow.fields[index++] = remaining.fields[i];
+			}
+		}
+
+		return joinedRow;
+	}
 }
diff --git a/flink-python/MANIFEST.in b/flink-python/MANIFEST.in
@@ -29,3 +29,4 @@ include README.md
 include pyflink/LICENSE
 include pyflink/NOTICE
 include pyflink/README.txt
+graft pyflink/proto
diff --git a/flink-python/pom.xml b/flink-python/pom.xml
@@ -23,8 +23,8 @@ under the License.
 	<modelVersion>4.0.0</modelVersion>
 
 	<parent>
-		<artifactId>flink-parent</artifactId>
 		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-parent</artifactId>
 		<version>1.10-SNAPSHOT</version>
 		<relativePath>..</relativePath>
 	</parent>
@@ -62,13 +62,32 @@ under the License.
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
+		<!-- Beam dependencies -->
+
+		<dependency>
+			<groupId>org.apache.beam</groupId>
+			<artifactId>beam-runners-java-fn-execution</artifactId>
+		</dependency>
+
+		<!-- Protobuf dependencies -->
+
+		<dependency>
+			<groupId>com.google.protobuf</groupId>
+			<artifactId>protobuf-java</artifactId>
+		</dependency>
 
 		<!-- Python API dependencies -->
 
 		<dependency>
 			<groupId>net.sf.py4j</groupId>
 			<artifactId>py4j</artifactId>
-			<version>${py4j.version}</version>
 		</dependency>
 		<dependency>
 			<groupId>net.razorvine</groupId>
@@ -83,6 +102,46 @@ under the License.
 		</dependency>
 	</dependencies>
 
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>net.sf.py4j</groupId>
+				<artifactId>py4j</artifactId>
+				<version>${py4j.version}</version>
+			</dependency>
+
+			<dependency>
+				<groupId>org.apache.beam</groupId>
+				<artifactId>beam-runners-java-fn-execution</artifactId>
+				<version>${beam.version}</version>
+			</dependency>
+
+			<dependency>
+				<groupId>com.fasterxml.jackson.core</groupId>
+				<artifactId>jackson-core</artifactId>
+				<version>2.9.9</version>
+			</dependency>
+
+			<dependency>
+				<groupId>com.fasterxml.jackson.core</groupId>
+				<artifactId>jackson-databind</artifactId>
+				<version>2.9.9</version>
+			</dependency>
+
+			<dependency>
+				<groupId>com.fasterxml.jackson.core</groupId>
+				<artifactId>jackson-annotations</artifactId>
+				<version>2.9.9</version>
+			</dependency>
+
+			<dependency>
+				<groupId>com.google.protobuf</groupId>
+				<artifactId>protobuf-java</artifactId>
+				<version>${protoc.version}</version>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
 	<build>
 		<plugins>
 			<plugin>
@@ -128,7 +187,6 @@ under the License.
 					</execution>
 				</executions>
 			</plugin>
-
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-shade-plugin</artifactId>
@@ -140,10 +198,15 @@ under the License.
 							<goal>shade</goal>
 						</goals>
 						<configuration>
+							<shadeTestJar>false</shadeTestJar>
 							<artifactSet>
 								<includes combine.children="append">
 									<include>net.razorvine:*</include>
 									<include>net.sf.py4j:*</include>
+									<include>org.apache.beam:*</include>
+									<include>com.fasterxml.jackson.core:*</include>
+									<include>joda-time:*</include>
+									<inculde>com.google.protobuf:*</inculde>
 								</includes>
 							</artifactSet>
 							<relocations combine.children="append">
@@ -155,11 +218,42 @@ under the License.
 									<pattern>net.razorvine</pattern>
 									<shadedPattern>org.apache.flink.api.python.shaded.net.razorvine</shadedPattern>
 								</relocation>
+								<relocation>
+									<pattern>com.fasterxml.jackson</pattern>
+									<shadedPattern>org.apache.flink.api.python.shaded.com.fasterxml.jackson</shadedPattern>
+								</relocation>
+								<relocation>
+									<pattern>org.joda.time</pattern>
+									<shadedPattern>org.apache.flink.api.python.shaded.org.joda.time</shadedPattern>
+								</relocation>
+								<relocation>
+									<pattern>com.google.protobuf</pattern>
+									<shadedPattern>org.apache.flink.api.python.shaded.com.google.protobuf</shadedPattern>
+								</relocation>
 							</relocations>
 						</configuration>
 					</execution>
 				</executions>
 			</plugin>
+			<plugin>
+				<groupId>com.github.os72</groupId>
+				<artifactId>protoc-jar-maven-plugin</artifactId>
+				<version>${protoc.version}</version>
+				<executions>
+					<execution>
+						<phase>generate-sources</phase>
+						<goals>
+							<goal>run</goal>
+						</goals>
+						<configuration>
+							<protocVersion>${protoc.version}</protocVersion>
+							<inputDirectories>
+								<include>pyflink/proto</include>
+							</inputDirectories>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
 		</plugins>
 	</build>
 </project>
diff --git a/flink-python/pyflink/fn_execution/__init__.py b/flink-python/pyflink/fn_execution/__init__.py
@@ -0,0 +1,17 @@
+################################################################################
+#  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.
+################################################################################