[FLINK-14581][python] Let python UDF execution no longer rely on the flink directory structure to support running python UDFs on yarn. #10061

WeiZhong94 · 2019-10-31T11:34:06Z

What is the purpose of the change

This pull request makes python UDF execution no longer rely on the flink directory structure to support running python UDFs on yarn.

Brief change log

add pyflink-udf-runner.sh, pyflink.zip, py4j-0.10.8.1-src.zip and cloudpickle-1.2.2-src.zip as resources to flink-python-{VERSION}.jar.
add RunnerEnvUtil to extract above files before running python udf worker.
rework pyflink-udf-runner.sh and relevant tests to adapt to the above changes.

Verifying this change

This change is already covered by existing tests, such as RunnerEnvUtilTest, test_process_mode_boot.py etc.

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, 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)

…flink directory structure to support running python UDFs on yarn.

flinkbot · 2019-10-31T11:37:46Z

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 99b5327 (Wed Dec 04 15:24:20 UTC 2019)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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 for a full explanation of the review process.

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

Bot commands

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

flinkbot · 2019-10-31T11:53:29Z

CI report:

6f081df : SUCCESS Build
df8e683 : CANCELED Build
0113b4e : UNKNOWN
fb22e17 : FAILURE Build
da49a93 : FAILURE Build
99b5327 : SUCCESS Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

hequn8128

@WeiZhong94 Thanks a lot for the pr. Good job! I left some comments below. Mainly include:

There are some code format problems. For example, the extractBasicDependenciesFromResource method in RunnerEnvUtil. It's better to declare each parameter in a new line. I only left the comment in one place and please check other places.
Maybe it's better to place the methods from RunnerEnvUtil into AbstractPythonFunctionRunner directly. The methods seem not util methods that can be used by other places.
It's better to make the python log path the same with taskmanager log path.

Best, Hequn

hequn8128 · 2019-11-05T08:24:56Z

flink-python/pom.xml

+			<resource>
+				<directory>lib</directory>
+				<includes>
+					<include>*.zip</include>


If we include these zip files in the jar, can we remove the current python folder in opt folder(i.e., build-target/opt/python)?

hequn8128 · 2019-11-05T08:48:36Z

flink-python/src/main/java/org/apache/flink/python/util/RunnerEnvUtil.java

+ */
+public class RunnerEnvUtil {
+
+	public static final String[] PYTHON_BASIC_DEPENDENCIES = new String[] {


public => private

remove new String[]

hequn8128 · 2019-11-05T09:49:18Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+	 * Python libraries and shell script extracted from resource of flink-python jar.
+	 * They are used to support running python udf worker in process mode.
+	 */
+	private transient List<File> pythonFiles;


The name is easy to be mixed with python files uploaded by users after we support dependency management. So maybe we can rename it to be more clear? How about renaming it to internalPythonFiles?

hequn8128 · 2019-11-05T10:21:52Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+					break;
+				}
+			}
+			Map<String, String> env = RunnerEnvUtil.appendEnvironmentVariable(System.getenv(),


Put System.getenv(), to a new line.

hequn8128 · 2019-11-05T11:46:02Z

flink-python/src/main/java/org/apache/flink/python/util/RunnerEnvUtil.java

+		if (systemEnv.get("FLINK_LOG_DIR") == null) {
+			if (systemEnv.get("LOG_DIRS") != null) {
+				// log directory of yarn mode
+				result.put("FLINK_LOG_DIR", systemEnv.get("LOG_DIRS").split(File.pathSeparator)[0]);
+			} else if (systemEnv.get(ConfigConstants.ENV_FLINK_LIB_DIR) != null) {
+				// log directory of standalone mode
+				File flinkHomeDir = new File(systemEnv.get(ConfigConstants.ENV_FLINK_LIB_DIR)).getParentFile();
+				result.put("FLINK_LOG_DIR", new File(flinkHomeDir, "log").getAbsolutePath());
+			} else if (systemEnv.get(ConfigConstants.ENV_FLINK_HOME_DIR) != null) {
+				// log directory of pyflink shell mode or user specified flink
+				result.put("FLINK_LOG_DIR",
+					new File(systemEnv.get(ConfigConstants.ENV_FLINK_HOME_DIR), "log").getAbsolutePath());
+			}
+		}


The log path can be same with the log path of taskmanager?

Currently using System.getProperty("log.file") (default value of the log path of taskmanager) here, after moving the environment creation into EnvironmentManager in #10017 we can pass the true log path to this method easily.

hequn8128 · 2019-11-05T11:52:28Z

flink-python/src/main/java/org/apache/flink/python/util/RunnerEnvUtil.java

+ * The basic environment does not include python part of Apache Beam.
+ * Users need to prepare it themselves.
+ */
+public class RunnerEnvUtil {


How about put these two methods into AbstractPythonFunctionRunner for now and move them into EnvironmentManager in #10017.

It seems the two methods are not util methods that can be used by other places.

Due to the extract method is reused by the PythonResourceExtractor, I only move the appendEnvironmentVariable into AbstractPythonFunctionRunner.

hequn8128 · 2019-11-05T12:06:06Z

flink-python/src/main/java/org/apache/flink/python/util/RunnerEnvUtil.java

+		"pyflink-udf-runner.sh"
+	};
+
+	public static List<File> extractBasicDependenciesFromResource(


Be careful with the format. You can take a look at the methods in KeyedStream.

public static List<File> extractBasicDependenciesFromResource( String tmpdir, ClassLoader classLoader, String prefix) throws IOException {

Also check other places.

WeiZhong94 · 2019-11-07T09:08:01Z

@hequn8128 Thanks for your review! I have addressed your comments in the latest commit, please take a look again :).

hequn8128

@WeiZhong94 Thanks a lot for the update. The code looks better. I left some further comments below.

hequn8128 · 2019-11-18T06:18:41Z

flink-python/src/main/java/org/apache/flink/client/python/PythonResourceExtractor.java

+	public static void main(String[] args) throws IOException {
+		String tmpdir = System.getProperty("java.io.tmpdir");
+
+		List<File> files = extractBasicDependenciesFromResource(tmpdir,


Pay attention to the format.

hequn8128 · 2019-11-18T06:30:08Z

flink-python/src/main/java/org/apache/flink/client/python/PythonResourceExtractor.java

+		List<File> files = extractBasicDependenciesFromResource(tmpdir,
+			PythonResourceExtractor.class.getClassLoader(), UUID.randomUUID().toString());
+
+		files.stream().filter(file -> file.getName().endsWith(".sh")).forEach(File::delete);


Could we avoid this. For example, we can pass a list to the extractBasicDependenciesFromResource method that the list contains no sh files.

hequn8128 · 2019-11-18T06:31:00Z

flink-python/src/main/java/org/apache/flink/client/python/PythonResourceExtractor.java

+
+		files.stream().filter(file -> file.getName().endsWith(".sh")).forEach(File::delete);
+
+		System.out.print(files.stream().filter(file -> !file.getName().endsWith(".sh"))


We can also remove the filter(file -> !file.getName().endsWith(".sh")) here.

hequn8128 · 2019-11-18T06:31:53Z

flink-python/src/main/java/org/apache/flink/client/python/PythonResourceExtractor.java

+import static org.apache.flink.python.util.ResourceUtil.extractBasicDependenciesFromResource;
+
+/**
+ * The program that extracts the internal python libraries and


Format the comment length to about 100 char length. It is common for java.

hequn8128 · 2019-11-18T07:55:32Z

flink-python/bin/pyflink-shell.sh

+EXTRACTOR="org.apache.flink.client.python.PythonResourceExtractor"
+PYFLINK_INTERNAL_LIB=`${JAVA_RUN} ${JVM_ARGS} -cp ${PYTHON_JAR_PATH} ${EXTRACTOR}`
+export PYTHONPATH="$PYFLINK_INTERNAL_LIB:$PYTHONPATH"
+export PYFLINK_INTERNAL_LIB


Also clean temp files of PYFLINK_INTERNAL_LIB after executing pyflink.shell. If python shell exits abnormally, the file would not be cleaned.

hequn8128 · 2019-11-18T08:06:50Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+			String tmpdir = tempDirs[rnd.nextInt(tempDirs.length)];
+			String prefix = UUID.randomUUID().toString() + "_";
+			try {
+				pythonInternalLibs = ResourceUtil.extractBasicDependenciesFromResource(


Pay attention to the format. Change to:

pythonInternalLibs = ResourceUtil.extractBasicDependenciesFromResource( tmpdir, this.getClass().getClassLoader(), prefix);

hequn8128 · 2019-11-18T08:26:26Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+				LOG.warn("Can not get the log directory from property log.file.", e);
+			}
+		}
+		return appendEnvironmentVariable(


Pay attention to the code format.

hequn8128 · 2019-11-18T08:28:12Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+			pythonDependencies, logDir);
+	}
+
+	private static Map<String, String> appendEnvironmentVariable(


Merge this method with the previous appendEnvironmentVariable?

hequn8128 · 2019-11-18T08:42:53Z

flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java

 /**
 * The util class help to prepare Python env and run the python process.
 */
 public final class PythonEnvUtils {


How about rename this Class to PythonDriverEnvUtils ? I was confused when seeing extractBasicDependenciesFromResource was also called here at first glance.

hequn8128 · 2019-11-18T08:57:16Z

flink-python/src/test/java/org/apache/flink/python/util/ResourceUtilTest.java

+		try {
+			ClassLoader classLoader = ResourceUtil.class.getClassLoader();
+			String prefix = "tmp_";
+			List<File> files = ResourceUtil.extractBasicDependenciesFromResource(


Pay attention to the code format.

…yflink-shell.sh

WeiZhong94 · 2019-11-18T12:25:02Z

@hequn8128 Thanks for your review again! I have updated this PR. Please take a look. :)

hequn8128

@WeiZhong94 We should also update the tests in PythonDriverEnvUtilsTest. See the test error in misc

WeiZhong94 · 2019-11-19T04:18:31Z

@hequn8128 Thanks for your reminder, I have fixed the test failure in the latest commit.

hequn8128 · 2019-11-19T06:15:53Z

@WeiZhong94 Thanks a lot for the update. Merging...

…flink directory structure to support running python UDFs on yarn. This closes apache#10061.

[FLINK-14581][python] Let python UDF execution no longer rely on the …

6f081df

…flink directory structure to support running python UDFs on yarn.

rmetzger added the review=description? label Oct 31, 2019

rmetzger added the component=API/Python label Oct 31, 2019

hequn8128 reviewed Nov 5, 2019

View reviewed changes

remove pyflink.zip in opt directory and fix code style.

df8e683

WeiZhong94 added 2 commits November 7, 2019 17:25

remove unnecessary methods

0113b4e

remove unused imports

fb22e17

hequn8128 reviewed Nov 18, 2019

View reviewed changes

refactor ResourceUtils, optimize code style and add delete logic in p…

da49a93

…yflink-shell.sh

hequn8128 reviewed Nov 19, 2019

View reviewed changes

fix test failure

99b5327

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Nov 19, 2019

[FLINK-14581][python] Let python UDF execution no longer rely on the …

adaf866

…flink directory structure to support running python UDFs on yarn. This closes apache#10061.

hequn8128 closed this in df9b757 Nov 19, 2019

dianfu mentioned this pull request Dec 4, 2019

[FLINK-15025][python][legal] Fix the license and notice issue of flink-python module #10391

Merged


		files.stream().filter(file -> file.getName().endsWith(".sh")).forEach(File::delete);

		System.out.print(files.stream().filter(file -> !file.getName().endsWith(".sh"))

[FLINK-14581][python] Let python UDF execution no longer rely on the flink directory structure to support running python UDFs on yarn. #10061

[FLINK-14581][python] Let python UDF execution no longer rely on the flink directory structure to support running python UDFs on yarn. #10061

Uh oh!

Conversation

WeiZhong94 commented Oct 31, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Oct 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Oct 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

WeiZhong94 commented Nov 7, 2019

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

WeiZhong94 commented Nov 18, 2019

Uh oh!

hequn8128 left a comment

Choose a reason for hiding this comment

Uh oh!

WeiZhong94 commented Nov 19, 2019

Uh oh!

hequn8128 commented Nov 19, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

flinkbot commented Oct 31, 2019 •

edited

Loading

flinkbot commented Oct 31, 2019 •

edited

Loading