You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2018/06/15 15:32:22 UTC
[1/3] flink git commit: [FLINK-9366] DistributedCache works with
Distributed File System
Repository: flink
Updated Branches:
refs/heads/master fc3ee68bb -> 4a0dc8273
[FLINK-9366] DistributedCache works with Distributed File System
This closes #6107
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/167be0a9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/167be0a9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/167be0a9
Branch: refs/heads/master
Commit: 167be0a91d6f5863ecff04c1e13dfe60b11ae3b1
Parents: fc3ee68
Author: Dawid Wysakowicz <dw...@apache.org>
Authored: Thu May 31 14:35:44 2018 +0200
Committer: Dawid Wysakowicz <dw...@apache.org>
Committed: Fri Jun 15 14:27:49 2018 +0200
----------------------------------------------------------------------
.../api/common/cache/DistributedCache.java | 23 +--
.../hdfstests/DistributedCacheDfsTest.java | 160 +++++++++++++++++++
.../flink/api/java/ExecutionEnvironment.java | 4 +-
.../flink/runtime/filecache/FileCache.java | 31 +++-
.../apache/flink/runtime/jobgraph/JobGraph.java | 52 +++++-
.../flink/runtime/jobgraph/JobGraphTest.java | 19 +++
.../flink/api/scala/ExecutionEnvironment.scala | 5 +-
.../environment/StreamExecutionEnvironment.java | 4 +-
.../api/scala/StreamExecutionEnvironment.scala | 16 +-
tools/maven/suppressions-core.xml | 2 +-
10 files changed, 284 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java b/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
index 9cceb83..4bb4ece 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
@@ -15,8 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.api.common.cache;
+import org.apache.flink.annotation.Public;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
import java.io.File;
import java.io.Serializable;
@@ -29,10 +33,6 @@ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import org.apache.flink.annotation.Public;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.fs.Path;
-
/**
* DistributedCache provides static methods to write the registered cache files into job configuration or decode
* them from job configuration. It also provides user access to the file locally.
@@ -40,6 +40,9 @@ import org.apache.flink.core.fs.Path;
@Public
public class DistributedCache {
+ /**
+ * Meta info about an entry in {@link DistributedCache}.
+ */
public static class DistributedCacheEntry implements Serializable {
public String filePath;
@@ -48,9 +51,9 @@ public class DistributedCache {
public byte[] blobKey;
- public DistributedCacheEntry(String filePath, Boolean isExecutable, byte[] blobKey, boolean isZipped){
- this.filePath=filePath;
- this.isExecutable=isExecutable;
+ public DistributedCacheEntry(String filePath, Boolean isExecutable, byte[] blobKey, boolean isZipped) {
+ this.filePath = filePath;
+ this.isExecutable = isExecutable;
this.blobKey = blobKey;
this.isZipped = isZipped;
}
@@ -104,13 +107,15 @@ public class DistributedCache {
// ------------------------------------------------------------------------
public static void writeFileInfoToConfig(String name, DistributedCacheEntry e, Configuration conf) {
- int num = conf.getInteger(CACHE_FILE_NUM,0) + 1;
+ int num = conf.getInteger(CACHE_FILE_NUM, 0) + 1;
conf.setInteger(CACHE_FILE_NUM, num);
conf.setString(CACHE_FILE_NAME + num, name);
conf.setString(CACHE_FILE_PATH + num, e.filePath);
conf.setBoolean(CACHE_FILE_EXE + num, e.isExecutable || new File(e.filePath).canExecute());
conf.setBoolean(CACHE_FILE_DIR + num, e.isZipped || new File(e.filePath).isDirectory());
- conf.setBytes(CACHE_FILE_BLOB_KEY + num, e.blobKey);
+ if (e.blobKey != null) {
+ conf.setBytes(CACHE_FILE_BLOB_KEY + num, e.blobKey);
+ }
}
public static Set<Entry<String, DistributedCacheEntry>> readFileInfoFromConfig(Configuration conf) {
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java
----------------------------------------------------------------------
diff --git a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java
new file mode 100644
index 0000000..5d6a30f
--- /dev/null
+++ b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/DistributedCacheDfsTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.hdfstests;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.util.MiniClusterResource;
+import org.apache.flink.util.NetUtils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for distributing files with {@link org.apache.flink.api.common.cache.DistributedCache} via HDFS.
+ */
+public class DistributedCacheDfsTest {
+
+ private static final String testFileContent = "Goethe - Faust: Der Tragoedie erster Teil\n" + "Prolog im Himmel.\n"
+ + "Der Herr. Die himmlischen Heerscharen. Nachher Mephistopheles. Die drei\n" + "Erzengel treten vor.\n"
+ + "RAPHAEL: Die Sonne toent, nach alter Weise, In Brudersphaeren Wettgesang,\n"
+ + "Und ihre vorgeschriebne Reise Vollendet sie mit Donnergang. Ihr Anblick\n"
+ + "gibt den Engeln Staerke, Wenn keiner Sie ergruenden mag; die unbegreiflich\n"
+ + "hohen Werke Sind herrlich wie am ersten Tag.\n"
+ + "GABRIEL: Und schnell und unbegreiflich schnelle Dreht sich umher der Erde\n"
+ + "Pracht; Es wechselt Paradieseshelle Mit tiefer, schauervoller Nacht. Es\n"
+ + "schaeumt das Meer in breiten Fluessen Am tiefen Grund der Felsen auf, Und\n"
+ + "Fels und Meer wird fortgerissen Im ewig schnellem Sphaerenlauf.\n"
+ + "MICHAEL: Und Stuerme brausen um die Wette Vom Meer aufs Land, vom Land\n"
+ + "aufs Meer, und bilden wuetend eine Kette Der tiefsten Wirkung rings umher.\n"
+ + "Da flammt ein blitzendes Verheeren Dem Pfade vor des Donnerschlags. Doch\n"
+ + "deine Boten, Herr, verehren Das sanfte Wandeln deines Tags.";
+
+ @ClassRule
+ public static final TemporaryFolder TEMP_FOLDER = new TemporaryFolder();
+
+ @ClassRule
+ public static final MiniClusterResource MINI_CLUSTER_RESOURCE = new MiniClusterResource(
+ new MiniClusterResource.MiniClusterResourceConfiguration(
+ new org.apache.flink.configuration.Configuration(),
+ 1,
+ 1));
+
+ private static MiniDFSCluster hdfsCluster;
+ private static Configuration conf = new Configuration();
+
+ private static Path testFile;
+ private static Path testDir;
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ File dataDir = TEMP_FOLDER.newFolder();
+
+ conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, dataDir.getAbsolutePath());
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(conf);
+ hdfsCluster = builder.build();
+
+ String hdfsURI = "hdfs://"
+ + NetUtils.hostAndPortToUrlString(hdfsCluster.getURI().getHost(), hdfsCluster.getNameNodePort())
+ + "/";
+
+ FileSystem dfs = FileSystem.get(new URI(hdfsURI));
+ testFile = writeFile(dfs, dfs.getHomeDirectory(), "testFile");
+
+ testDir = new Path(dfs.getHomeDirectory(), "testDir");
+ dfs.mkdirs(testDir);
+ writeFile(dfs, testDir, "testFile1");
+ writeFile(dfs, testDir, "testFile2");
+ }
+
+ private static Path writeFile(FileSystem dfs, Path rootDir, String fileName) throws IOException {
+ Path file = new Path(rootDir, fileName);
+ try (
+ DataOutputStream outStream = new DataOutputStream(dfs.create(file,
+ FileSystem.WriteMode.OVERWRITE))) {
+ outStream.writeUTF(testFileContent);
+ }
+ return file;
+ }
+
+ @AfterClass
+ public static void teardown() {
+ hdfsCluster.shutdown();
+ }
+
+ @Test
+ public void testDistributeFileViaDFS() throws Exception {
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+
+ env.registerCachedFile(testFile.toString(), "test_data", false);
+ env.registerCachedFile(testDir.toString(), "test_dir", false);
+
+ env.fromElements(1)
+ .map(new TestMapFunction())
+ .print();
+
+ env.execute("Distributed Cache Via Blob Test Program");
+ }
+
+ private static class TestMapFunction extends RichMapFunction<Integer, String> {
+
+ private static final long serialVersionUID = -3917258280687242969L;
+
+ @Override
+ public String map(Integer value) throws Exception {
+ final Path actualFile = new Path(getRuntimeContext().getDistributedCache().getFile("test_data").toURI());
+
+ Path path = new Path(actualFile.toUri());
+ assertFalse(path.getFileSystem().isDistributedFS());
+
+ DataInputStream in = new DataInputStream(actualFile.getFileSystem().open(actualFile));
+ String contents = in.readUTF();
+
+ assertEquals(testFileContent, contents);
+
+ final Path actualDir = new Path(getRuntimeContext().getDistributedCache().getFile("test_dir").toURI());
+ FileStatus fileStatus = actualDir.getFileSystem().getFileStatus(actualDir);
+ assertTrue(fileStatus.isDir());
+ FileStatus[] fileStatuses = actualDir.getFileSystem().listStatus(actualDir);
+ assertEquals(2, fileStatuses.length);
+
+ return contents;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index 8206220..2bed411 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -844,7 +844,7 @@ public abstract class ExecutionEnvironment {
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a distributed file system.
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file system.
* The runtime will copy the files temporarily to a local cache, if needed.
*
* <p>The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs via
@@ -862,7 +862,7 @@ public abstract class ExecutionEnvironment {
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a distributed file system.
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file system.
* The runtime will copy the files temporarily to a local cache, if needed.
*
* <p>The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs via
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
index 9021751..fe6c6bc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
@@ -204,7 +204,12 @@ public class FileCache {
Path target = new Path(tempDirToUse.getAbsolutePath() + "/" + sourceFile);
// kick off the copying
- Callable<Path> cp = new CopyFromBlobProcess(entry, jobID, blobService, target);
+ Callable<Path> cp;
+ if (entry.blobKey != null) {
+ cp = new CopyFromBlobProcess(entry, jobID, blobService, target);
+ } else {
+ cp = new CopyFromDFSProcess(entry, target);
+ }
FutureTask<Path> copyTask = new FutureTask<>(cp);
executorService.submit(copyTask);
@@ -303,6 +308,30 @@ public class FileCache {
}
/**
+ * Asynchronous file copy process.
+ */
+ private static class CopyFromDFSProcess implements Callable<Path> {
+
+ private final Path filePath;
+ private final Path cachedPath;
+ private boolean executable;
+
+ public CopyFromDFSProcess(DistributedCacheEntry e, Path cachedPath) {
+ this.filePath = new Path(e.filePath);
+ this.executable = e.isExecutable;
+ this.cachedPath = cachedPath;
+ }
+
+ @Override
+ public Path call() throws IOException {
+ // let exceptions propagate. we can retrieve them later from
+ // the future and report them upon access to the result
+ FileUtils.copy(filePath, cachedPath, this.executable);
+ return cachedPath;
+ }
+ }
+
+ /**
* If no task is using this file after 5 seconds, clear it.
*/
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index bc1a795..061cabe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -37,6 +37,7 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -580,11 +581,54 @@ public class JobGraph implements Serializable {
return "JobGraph(jobId: " + jobID + ")";
}
- public void uploadUserArtifacts(InetSocketAddress blobServerAddress, Configuration clientConfig) throws IOException {
- if (!userArtifacts.isEmpty()) {
- try (BlobClient blobClient = new BlobClient(blobServerAddress, clientConfig)) {
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
+ /**
+ * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
+ * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
+ *
+ * @param blobServerAddress of the blob server to upload the files to
+ * @param blobClientConfig the blob client configuration
+ * @throws IOException Thrown, if the file upload to the Blob server failed.
+ */
+ public void uploadUserArtifacts(
+ InetSocketAddress blobServerAddress,
+ Configuration blobClientConfig) throws IOException {
+
+ Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
+ Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
+
+ for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
+ Path filePath = new Path(userArtifact.getValue().filePath);
+
+ try {
+ if (filePath.getFileSystem().isDistributedFS()) {
+ distributeViaDFS.add(userArtifact);
+ } else {
+ uploadToBlobServer.add(userArtifact);
+ }
+
+ } catch (IOException ex) {
+ distributeViaDFS.add(userArtifact);
+ }
+ }
+
+ uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
+ for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ DistributedCache.writeFileInfoToConfig(
+ userArtifact.getKey(),
+ userArtifact.getValue(),
+ jobConfiguration
+ );
+ }
+ }
+
+ private void uploadViaBlob(
+ InetSocketAddress blobServerAddress,
+ Configuration clientConfig,
+ Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer) throws IOException {
+ if (!uploadToBlobServer.isEmpty()) {
+ try (BlobClient blobClient = new BlobClient(blobServerAddress, clientConfig)) {
+ for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : uploadToBlobServer) {
final PermanentBlobKey key = blobClient.uploadFile(jobID,
new Path(userArtifact.getValue().filePath));
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
index 9f06c6a..8815acb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
@@ -20,9 +20,12 @@ package org.apache.flink.runtime.jobgraph;
import static org.junit.Assert.*;
+import java.net.InetSocketAddress;
import java.util.List;
import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.junit.Test;
@@ -258,6 +261,22 @@ public class JobGraphTest {
fail(e.getMessage());
}
}
+
+ @Test
+ public void testConfiguringDistributedCache() throws Exception {
+ JobGraph testJob = new JobGraph("Test job");
+ testJob.addUserArtifact("dfsFile", new DistributedCache.DistributedCacheEntry("hdfs://tmp/file", false));
+
+ //it should never try to connect to that address
+ testJob.uploadUserArtifacts(new InetSocketAddress("localhost", 1111), new Configuration());
+
+ Configuration jobConfiguration = testJob.getJobConfiguration();
+ assertEquals(1, jobConfiguration.getInteger("DISTRIBUTED_CACHE_FILE_NUM", -1));
+ assertFalse(jobConfiguration.getBoolean("DISTRIBUTED_CACHE_FILE_DIR_1", true));
+ assertEquals("dfsFile", jobConfiguration.getString("DISTRIBUTED_CACHE_FILE_NAME_1", ""));
+ assertEquals("hdfs://tmp/file", jobConfiguration.getString("DISTRIBUTED_CACHE_FILE_PATH_1", ""));
+ assertFalse(jobConfiguration.getBoolean("DISTRIBUTED_CACHE_FILE_EXE_1", true));
+ }
private static final void assertBefore(JobVertex v1, JobVertex v2, List<JobVertex> list) {
boolean seenFirst = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index 9f53123..00d2d77 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -493,9 +493,8 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it),
- * or files in a distributed file system.
- * The runtime will copy the files temporarily to a local cache, if needed.
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file
+ * system. The runtime will copy the files temporarily to a local cache, if needed.
*
* The [[org.apache.flink.api.common.functions.RuntimeContext]] can be obtained inside UDFs
* via
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index f35434f..d7ca7e0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -1804,7 +1804,7 @@ public abstract class StreamExecutionEnvironment {
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a distributed file system.
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file system.
* The runtime will copy the files temporarily to a local cache, if needed.
*
* <p>The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs via
@@ -1822,7 +1822,7 @@ public abstract class StreamExecutionEnvironment {
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a distributed file system.
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file system.
* The runtime will copy the files temporarily to a local cache, if needed.
*
* <p>The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs via
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index 9aac431..6a67d63 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -690,14 +690,12 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
f
}
-
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a
- * distributed file system. The runtime will copy the files temporarily to a local cache,
- * if needed.
- * <p>
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file
+ * system. The runtime will copy the files temporarily to a local cache, if needed.
+ *
* The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs
* via {@link org.apache.flink.api.common.functions.RichFunction#getRuntimeContext()} and
* provides access {@link org.apache.flink.api.common.cache.DistributedCache} via
@@ -711,14 +709,12 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
javaEnv.registerCachedFile(filePath, name)
}
-
/**
* Registers a file at the distributed cache under the given name. The file will be accessible
* from any user-defined function in the (distributed) runtime under a local path. Files
- * may be local files (as long as all relevant workers have access to it), or files in a
- * distributed file system. The runtime will copy the files temporarily to a local cache,
- * if needed.
- * <p>
+ * may be local files (which will be distributed via BlobServer), or files in a distributed file
+ * system. The runtime will copy the files temporarily to a local cache, if needed.
+ *
* The {@link org.apache.flink.api.common.functions.RuntimeContext} can be obtained inside UDFs
* via {@link org.apache.flink.api.common.functions.RichFunction#getRuntimeContext()} and
* provides access {@link org.apache.flink.api.common.cache.DistributedCache} via
http://git-wip-us.apache.org/repos/asf/flink/blob/167be0a9/tools/maven/suppressions-core.xml
----------------------------------------------------------------------
diff --git a/tools/maven/suppressions-core.xml b/tools/maven/suppressions-core.xml
index a2cd207..db5d16a 100644
--- a/tools/maven/suppressions-core.xml
+++ b/tools/maven/suppressions-core.xml
@@ -76,7 +76,7 @@ under the License.
checks="AvoidStarImport|ArrayTypeStyle|Regexp"/>
<suppress
- files="(.*)api[/\\]common[/\\](cache|distributions|restartstrategy)[/\\](.*)"
+ files="(.*)api[/\\]common[/\\](distributions|restartstrategy)[/\\](.*)"
checks="NewlineAtEndOfFile|RegexpSingleline|TodoComment|RedundantImport|ImportOrder|RedundantModifier|JavadocMethod|JavadocParagraph|JavadocType|JavadocStyle|PackageName|TypeNameCheck|ConstantNameCheck|StaticVariableNameCheck|MemberNameCheck|MethodNameCheck|ParameterName|LocalFinalVariableName|LocalVariableName|LeftCurly|UpperEll|FallThrough|reliefPattern|SimplifyBooleanExpression|EmptyStatement|ModifierOrder|EmptyLineSeparator|WhitespaceAround|WhitespaceAfter|NoWhitespaceAfter|NoWhitespaceBefore|OperatorWrap|ParenPad"/>
<suppress
[2/3] flink git commit: [FLINK-9579][CEP]Remove unneeded clear on
elementQueueState
Posted by dw...@apache.org.
[FLINK-9579][CEP]Remove unneeded clear on elementQueueState
This closes #6162
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2fdc768e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2fdc768e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2fdc768e
Branch: refs/heads/master
Commit: 2fdc768edd7cd88ab4584f7d9072640ba77e1678
Parents: 167be0a
Author: minwenjun <mi...@didichuxing.com>
Authored: Thu Jun 14 14:20:42 2018 +0800
Committer: Dawid Wysakowicz <dw...@apache.org>
Committed: Fri Jun 15 14:34:30 2018 +0200
----------------------------------------------------------------------
.../flink/cep/operator/AbstractKeyedCEPPatternOperator.java | 6 ------
1 file changed, 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2fdc768e/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index 1e482c3..7e3e6f3 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -293,9 +293,6 @@ public abstract class AbstractKeyedCEPPatternOperator<IN, KEY, OUT, F extends Fu
advanceTime(nfaState, timerService.currentWatermark());
// STEP 4
- if (sortedTimestamps.isEmpty()) {
- elementQueueState.clear();
- }
updateNFA(nfaState);
if (!sortedTimestamps.isEmpty() || !partialMatches.isEmpty()) {
@@ -337,9 +334,6 @@ public abstract class AbstractKeyedCEPPatternOperator<IN, KEY, OUT, F extends Fu
}
// STEP 3
- if (sortedTimestamps.isEmpty()) {
- elementQueueState.clear();
- }
updateNFA(nfa);
}
[3/3] flink git commit: [FLINK-9551][DOCS]FlinkCEP Scala Combining
Patterns table has a missing pattern
Posted by dw...@apache.org.
[FLINK-9551][DOCS]FlinkCEP Scala Combining Patterns table has a missing pattern
This closes #6139
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4a0dc827
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4a0dc827
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4a0dc827
Branch: refs/heads/master
Commit: 4a0dc8273f422dbf28e105760b52bdcc3cffb8e7
Parents: 2fdc768
Author: Yadan.JS <y_...@yahoo.com>
Authored: Thu Jun 7 18:07:57 2018 -0400
Committer: Dawid Wysakowicz <dw...@apache.org>
Committed: Fri Jun 15 15:26:37 2018 +0200
----------------------------------------------------------------------
docs/dev/libs/cep.md | 13 ++++++++++++-
1 file changed, 12 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4a0dc827/docs/dev/libs/cep.md
----------------------------------------------------------------------
diff --git a/docs/dev/libs/cep.md b/docs/dev/libs/cep.md
index 976be34..cdcebdc 100644
--- a/docs/dev/libs/cep.md
+++ b/docs/dev/libs/cep.md
@@ -1117,7 +1117,7 @@ pattern.within(Time.seconds(10));
</thead>
<tbody>
<tr>
- <td><strong>begin()</strong></td>
+ <td><strong>begin(#name)</strong></td>
<td>
<p>Defines a starting pattern:</p>
{% highlight scala %}
@@ -1125,6 +1125,17 @@ val start = Pattern.begin[Event]("start")
{% endhighlight %}
</td>
</tr>
+ <tr>
+ <td><strong>begin(#pattern_sequence)</strong></td>
+ <td>
+ <p>Defines a starting pattern:</p>
+{% highlight scala %}
+val start = Pattern.begin(
+ Pattern.begin[Event]("start").where(...).followedBy("middle").where(...)
+)
+{% endhighlight %}
+ </td>
+ </tr>
<tr>
<td><strong>next(#name)</strong></td>
<td>