You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2021/01/27 06:55:42 UTC
[flink] branch release-1.12 updated: [FLINK-21140][python] Extract
zip file dependencies before adding to PYTHONPATH
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.12 by this push:
new fa5619c [FLINK-21140][python] Extract zip file dependencies before adding to PYTHONPATH
fa5619c is described below
commit fa5619c5a0d9589d990dee8a2e35c39ae06d0d35
Author: Dian Fu <di...@apache.org>
AuthorDate: Tue Jan 26 11:18:17 2021 +0800
[FLINK-21140][python] Extract zip file dependencies before adding to PYTHONPATH
This closes #14755.
---
docs/_includes/generated/python_configuration.html | 2 +-
.../apache/flink/client/cli/CliFrontendParser.java | 4 +--
.../apache/flink/client/python/PythonEnvUtils.java | 10 ++++++++
.../org/apache/flink/python/PythonOptions.java | 9 +++----
.../env/beam/ProcessPythonEnvironmentManager.java | 15 +++++++++++
.../flink/client/python/PythonEnvUtilsTest.java | 17 ++++++++++---
.../beam/ProcessPythonEnvironmentManagerTest.java | 29 +++++++++-------------
7 files changed, 57 insertions(+), 29 deletions(-)
diff --git a/docs/_includes/generated/python_configuration.html b/docs/_includes/generated/python_configuration.html
index 16b76e2..88e25cc 100644
--- a/docs/_includes/generated/python_configuration.html
+++ b/docs/_includes/generated/python_configuration.html
@@ -30,7 +30,7 @@
<td><h5>python.files</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Attach custom python files for job. These files will be added to the PYTHONPATH of both the local client and the remote python UDF worker. The standard python resource file suffixes such as .py/.egg/.zip or directory are all supported. Comma (',') could be used as the separator to specify multiple files. The option is equivalent to the command line option "-pyfs". </td>
+ <td>Attach custom python files for job. The standard python resource file suffixes such as .py/.egg/.zip or directory are all supported. These files will be added to the PYTHONPATH of both the local client and the remote python UDF worker. Files suffixed with .zip will be extracted and added to PYTHONPATH. Comma (',') could be used as the separator to specify multiple files. The option is equivalent to the command line option "-pyfs". </td>
</tr>
<tr>
<td><h5>python.fn-execution.arrow.batch.size</h5></td>
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index 0e7fd23..ce4acc9 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -191,9 +191,9 @@ public class CliFrontendParser {
"pyfs",
"pyFiles",
true,
- "Attach custom python files for job. "
+ "Attach custom python files for job. The standard python resource file suffixes such as .py/.egg/.zip or directory are all supported. "
+ "These files will be added to the PYTHONPATH of both the local client and the remote python UDF worker. "
- + "The standard python resource file suffixes such as .py/.egg/.zip or directory are all supported. "
+ + "Files suffixed with .zip will be extracted and added to PYTHONPATH. "
+ "Comma (',') could be used as the separator to specify multiple files "
+ "(e.g.: --pyFiles file:///tmp/myresource.zip,hdfs:///$namenode_address/myresource2.zip).");
diff --git a/flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java b/flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java
index 758e39b..64be8a2 100644
--- a/flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java
+++ b/flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java
@@ -242,6 +242,16 @@ final class PythonEnvUtils {
&& sourceFileName.endsWith(".py")) {
// add the parent directory of .py file itself to PYTHONPATH
pythonPathList.add(targetPath.getParent().toString());
+ } else if (Files.isRegularFile(Paths.get(targetPath.toString()).toRealPath())
+ && sourceFileName.endsWith(".zip")) {
+ // expand the zip file and add the root directory to PYTHONPATH
+ // as not all zip files are importable
+ Path targetDirectory =
+ new Path(
+ targetPath.getParent(),
+ sourceFileName.substring(0, sourceFileName.lastIndexOf(".")));
+ FileUtils.expandDirectory(targetPath, targetDirectory);
+ pythonPathList.add(targetDirectory.toString());
} else {
pythonPathList.add(targetPath.toString());
}
diff --git a/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java b/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java
index fd95c7d..bcde0ab 100644
--- a/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java
+++ b/flink-python/src/main/java/org/apache/flink/python/PythonOptions.java
@@ -70,11 +70,10 @@ public class PythonOptions {
.stringType()
.noDefaultValue()
.withDescription(
- "Attach custom python files for job. These files will "
- + "be added to the PYTHONPATH of both the local client and the remote python UDF "
- + "worker. The standard python resource file suffixes such as .py/.egg/.zip or "
- + "directory are all supported. Comma (',') could be used as the separator to specify "
- + "multiple files. The option is equivalent to the command line option \"-pyfs\". ");
+ "Attach custom python files for job. The standard python resource file suffixes such as .py/.egg/.zip or "
+ + "directory are all supported. These files will be added to the PYTHONPATH of both the local "
+ + "client and the remote python UDF worker. Files suffixed with .zip will be extracted and added to PYTHONPATH. "
+ + "Comma (',') could be used as the separator to specify multiple files. The option is equivalent to the command line option \"-pyfs\". ");
public static final ConfigOption<String> PYTHON_REQUIREMENTS =
ConfigOptions.key("python.requirements")
diff --git a/flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java b/flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
index e76b173..054f71f 100644
--- a/flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
+++ b/flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
@@ -287,6 +287,21 @@ public final class ProcessPythonEnvironmentManager implements PythonEnvironmentM
// If the python file is file with suffix .py, add the parent directory to
// PYTHONPATH.
pythonPath = String.join(File.separator, filesDirectory, distributedCacheFileName);
+ } else if (pythonFile.isFile() && originFileName.endsWith(".zip")) {
+ // Expand the zip file and add the root directory to PYTHONPATH
+ // as not all zip files are importable
+ org.apache.flink.core.fs.Path targetDirectory =
+ new org.apache.flink.core.fs.Path(
+ filesDirectory,
+ String.join(
+ File.separator,
+ distributedCacheFileName,
+ originFileName.substring(
+ 0, originFileName.lastIndexOf("."))));
+ FileUtils.expandDirectory(
+ new org.apache.flink.core.fs.Path(pythonFile.getAbsolutePath()),
+ targetDirectory);
+ pythonPath = targetDirectory.toString();
} else {
pythonPath =
String.join(
diff --git a/flink-python/src/test/java/org/apache/flink/client/python/PythonEnvUtilsTest.java b/flink-python/src/test/java/org/apache/flink/client/python/PythonEnvUtilsTest.java
index 43c6b46..6105d5c 100644
--- a/flink-python/src/test/java/org/apache/flink/client/python/PythonEnvUtilsTest.java
+++ b/flink-python/src/test/java/org/apache/flink/client/python/PythonEnvUtilsTest.java
@@ -24,6 +24,8 @@ import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.OperatingSystem;
+import org.apache.commons.compress.archivers.zip.ZipArchiveEntry;
+import org.apache.commons.compress.archivers.zip.ZipArchiveOutputStream;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
@@ -31,6 +33,7 @@ import org.junit.Before;
import org.junit.Test;
import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
@@ -76,10 +79,16 @@ public class PythonEnvUtilsTest {
File relativeFile =
new File(tmpDirPath + File.separator + "subdir" + File.separator + "b.py");
File schemeFile =
- new File(tmpDirPath + File.separator + "subdir" + File.separator + "c.zip");
+ new File(tmpDirPath + File.separator + "subdir" + File.separator + "c.egg");
// The files must actually exist
- zipFile.createNewFile();
+ try (ZipArchiveOutputStream zipOut =
+ new ZipArchiveOutputStream(new FileOutputStream(zipFile))) {
+ ZipArchiveEntry entry = new ZipArchiveEntry("zipDir" + "/zipfile0");
+ zipOut.putArchiveEntry(entry);
+ zipOut.write(new byte[] {1, 1, 1, 1, 1});
+ zipOut.closeArchiveEntry();
+ }
dirFile.mkdir();
subdirFile.mkdir();
relativeFile.createNewFile();
@@ -112,10 +121,10 @@ public class PythonEnvUtilsTest {
String base = replaceUUID(env.tempDirectory);
Set<String> expectedPythonPaths = new HashSet<>();
- expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "a.zip"));
+ expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "a"));
expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "module_dir"));
expectedPythonPaths.add(String.join(File.separator, base, "{uuid}"));
- expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "c.zip"));
+ expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "c.egg"));
Set<String> actualPaths =
Arrays.stream(env.pythonPath.split(File.pathSeparator))
diff --git a/flink-python/src/test/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManagerTest.java b/flink-python/src/test/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManagerTest.java
index 2be2933..5454a5a 100644
--- a/flink-python/src/test/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManagerTest.java
+++ b/flink-python/src/test/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManagerTest.java
@@ -162,9 +162,9 @@ public class ProcessPythonEnvironmentManagerTest {
public void testPythonFiles() throws Exception {
// use LinkedHashMap to preserve the path order in environment variable
Map<String, String> pythonFiles = new LinkedHashMap<>();
- pythonFiles.put(String.join(File.separator, tmpDir, "file0"), "test_file1.py");
- pythonFiles.put(String.join(File.separator, tmpDir, "file1"), "test_file2.zip");
- pythonFiles.put(String.join(File.separator, tmpDir, "file2"), "test_file3.egg");
+ pythonFiles.put(String.join(File.separator, tmpDir, "zip0"), "test_zip.zip");
+ pythonFiles.put(String.join(File.separator, tmpDir, "file1"), "test_file1.py");
+ pythonFiles.put(String.join(File.separator, tmpDir, "file2"), "test_file2.egg");
pythonFiles.put(String.join(File.separator, tmpDir, "dir0"), "test_dir");
PythonDependencyInfo dependencyInfo =
new PythonDependencyInfo(pythonFiles, null, null, new HashMap<>(), "python");
@@ -178,42 +178,37 @@ public class ProcessPythonEnvironmentManagerTest {
String baseDir = environmentManager.getBaseDirectory();
String[] expectedUserPythonPaths =
new String[] {
- String.join(File.separator, baseDir, PYTHON_FILES_DIR, "file0"),
- String.join(
- File.separator,
- baseDir,
- PYTHON_FILES_DIR,
- "file1",
- "test_file2.zip"),
+ String.join(File.separator, baseDir, PYTHON_FILES_DIR, "zip0", "test_zip"),
+ String.join(File.separator, baseDir, PYTHON_FILES_DIR, "file1"),
String.join(
File.separator,
baseDir,
PYTHON_FILES_DIR,
"file2",
- "test_file3.egg"),
+ "test_file2.egg"),
String.join(File.separator, baseDir, PYTHON_FILES_DIR, "dir0", "test_dir")
};
String expectedPythonPath = String.join(File.pathSeparator, expectedUserPythonPaths);
assertEquals(expectedPythonPath, environmentVariable.get("PYTHONPATH"));
assertFileEquals(
- new File(String.join(File.separator, tmpDir, "file0")),
+ new File(String.join(File.separator, tmpDir, "file1")),
new File(
String.join(
File.separator,
baseDir,
PYTHON_FILES_DIR,
- "file0",
+ "file1",
"test_file1.py")));
assertFileEquals(
- new File(String.join(File.separator, tmpDir, "file1")),
+ new File(String.join(File.separator, tmpDir, "zipExpected0")),
new File(
String.join(
File.separator,
baseDir,
PYTHON_FILES_DIR,
- "file1",
- "test_file2.zip")));
+ "zip0",
+ "test_zip")));
assertFileEquals(
new File(String.join(File.separator, tmpDir, "file2")),
new File(
@@ -222,7 +217,7 @@ public class ProcessPythonEnvironmentManagerTest {
baseDir,
PYTHON_FILES_DIR,
"file2",
- "test_file3.egg")));
+ "test_file2.egg")));
assertFileEquals(
new File(String.join(File.separator, tmpDir, "dir0")),
new File(