Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
3655dee5
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
3655dee5
编写于
4月 09, 2017
作者:
Z
Zohar Mizrahi
提交者:
zentol
5月 04, 2017
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-6177] Add support for "Distributed Cache" in streaming applications
This closes #3741.
上级
127f1df6
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
130 addition
and
32 deletion
+130
-32
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
...streaming/api/environment/StreamExecutionEnvironment.java
+48
-0
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
...flink/streaming/api/graph/StreamingJobGraphGenerator.java
+6
-0
flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
...link/streaming/api/scala/StreamExecutionEnvironment.scala
+48
-0
flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
...che/flink/test/distributedCache/DistributedCacheTest.java
+28
-32
未找到文件。
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
浏览文件 @
3655dee5
...
...
@@ -33,6 +33,7 @@ import org.apache.flink.annotation.PublicEvolving;
import
org.apache.flink.api.common.ExecutionConfig
;
import
org.apache.flink.api.common.InvalidProgramException
;
import
org.apache.flink.api.common.JobExecutionResult
;
import
org.apache.flink.api.common.cache.DistributedCache
;
import
org.apache.flink.api.common.functions.InvalidTypesException
;
import
org.apache.flink.api.common.functions.StoppableFunction
;
import
org.apache.flink.api.common.io.FileInputFormat
;
...
...
@@ -46,6 +47,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import
org.apache.flink.api.java.ClosureCleaner
;
import
org.apache.flink.api.java.ExecutionEnvironment
;
import
org.apache.flink.api.java.io.TextInputFormat
;
import
org.apache.flink.api.java.tuple.Tuple2
;
import
org.apache.flink.api.java.tuple.Tuple3
;
import
org.apache.flink.api.java.typeutils.MissingTypeInfo
;
import
org.apache.flink.api.java.typeutils.PojoTypeInfo
;
...
...
@@ -136,6 +138,8 @@ public abstract class StreamExecutionEnvironment {
/** The time characteristic used by the data streams. */
private
TimeCharacteristic
timeCharacteristic
=
DEFAULT_TIME_CHARACTERISTIC
;
protected
final
List
<
Tuple2
<
String
,
DistributedCache
.
DistributedCacheEntry
>>
cacheFile
=
new
ArrayList
<>();
// --------------------------------------------------------------------------------------------
// Constructor and Properties
...
...
@@ -148,6 +152,13 @@ public abstract class StreamExecutionEnvironment {
return
config
;
}
/**
* Get the list of cached files that were registered for distribution among the task managers.
*/
public
List
<
Tuple2
<
String
,
DistributedCache
.
DistributedCacheEntry
>>
getCachedFiles
()
{
return
cacheFile
;
}
/**
* Sets the parallelism for operations executed through this environment.
* Setting a parallelism of x here will cause all operators (such as map,
...
...
@@ -1774,4 +1785,41 @@ public abstract class StreamExecutionEnvironment {
protected
static
void
resetContextEnvironment
()
{
contextEnvironmentFactory
=
null
;
}
/**
* 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>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
* {@link org.apache.flink.api.common.functions.RuntimeContext#getDistributedCache()}.
*
* @param filePath The path of the file, as a URI (e.g. "file:///some/path" or "hdfs://host:port/and/path")
* @param name The name under which the file is registered.
*/
public
void
registerCachedFile
(
String
filePath
,
String
name
)
{
registerCachedFile
(
filePath
,
name
,
false
);
}
/**
* 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>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
* {@link org.apache.flink.api.common.functions.RuntimeContext#getDistributedCache()}.
*
* @param filePath The path of the file, as a URI (e.g. "file:///some/path" or "hdfs://host:port/and/path")
* @param name The name under which the file is registered.
* @param executable flag indicating whether the file should be executable
*/
public
void
registerCachedFile
(
String
filePath
,
String
name
,
boolean
executable
)
{
this
.
cacheFile
.
add
(
new
Tuple2
<>(
name
,
new
DistributedCache
.
DistributedCacheEntry
(
filePath
,
executable
)));
}
}
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
浏览文件 @
3655dee5
...
...
@@ -29,6 +29,7 @@ import java.util.Map.Entry;
import
org.apache.commons.lang3.StringUtils
;
import
org.apache.flink.annotation.Internal
;
import
org.apache.flink.api.common.cache.DistributedCache
;
import
org.apache.flink.api.common.functions.Function
;
import
org.apache.flink.api.common.operators.ResourceSpec
;
import
org.apache.flink.api.common.operators.util.UserCodeObjectWrapper
;
...
...
@@ -151,6 +152,11 @@ public class StreamingJobGraphGenerator {
configureCheckpointing
();
// add registered cache file into job configuration
for
(
Tuple2
<
String
,
DistributedCache
.
DistributedCacheEntry
>
e
:
streamGraph
.
getEnvironment
().
getCachedFiles
())
{
DistributedCache
.
writeFileInfoToConfig
(
e
.
f0
,
e
.
f1
,
jobGraph
.
getJobConfiguration
());
}
// set the ExecutionConfig last when it has been finalized
try
{
jobGraph
.
setExecutionConfig
(
streamGraph
.
getExecutionConfig
());
...
...
flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
浏览文件 @
3655dee5
...
...
@@ -49,6 +49,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
*/
def
getConfig
=
javaEnv
.
getConfig
/**
* Gets cache files.
*/
def
getCachedFiles
=
javaEnv
.
getCachedFiles
/**
* Sets the parallelism for operations executed through this environment.
* Setting a parallelism of x here will cause all operators (such as join, map, reduce) to run
...
...
@@ -668,6 +673,49 @@ 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>
* 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
* {@link org.apache.flink.api.common.functions.RuntimeContext#getDistributedCache()}.
*
* @param filePath The path of the file, as a URI (e.g. "file:///some/path" or
* "hdfs://host:port/and/path")
* @param name The name under which the file is registered.
*/
def
registerCachedFile
(
filePath
:
String
,
name
:
String
)
:
Unit
=
{
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>
* 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
* {@link org.apache.flink.api.common.functions.RuntimeContext#getDistributedCache()}.
*
* @param filePath The path of the file, as a URI (e.g. "file:///some/path" or
* "hdfs://host:port/and/path")
* @param name The name under which the file is registered.
* @param executable flag indicating whether the file should be executable
*/
def
registerCachedFile
(
filePath
:
String
,
name
:
String
,
executable
:
Boolean
)
:
Unit
=
{
javaEnv
.
registerCachedFile
(
filePath
,
name
,
executable
)
}
}
object
StreamExecutionEnvironment
{
...
...
flink-tests/src/test/java/org/apache/flink/test/distributedCache/DistributedCacheTest.java
浏览文件 @
3655dee5
...
...
@@ -15,29 +15,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
org.apache.flink.test.distributedCache
;
import
java.io.BufferedReader
;
import
java.io.File
;
import
java.io.IOException
;
import
java.io.FileNotFoundException
;
import
java.io.FileReader
;
import
java.util.HashSet
;
import
java.util.List
;
import
java.util.Set
;
package
org.apache.flink.test.distributedCache
;
import
org.apache.flink.api.common.functions.RichFlatMapFunction
;
import
org.apache.flink.api.java.ExecutionEnvironment
;
import
org.apache.flink.api.java.tuple.Tuple1
;
import
org.apache.flink.configuration.Configuration
;
import
org.apache.flink.test.util.JavaProgramTestBase
;
import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
;
import
org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
;
import
org.apache.flink.util.Collector
;
import
org.junit.Test
;
import
static
org
.
junit
.
Assert
.
assertTrue
;
import
java.io.*
;
import
java.util.*
;
/**
* Tests the distributed cache by comparing a text file with a distributed copy.
*/
public
class
DistributedCacheTest
extends
JavaProgramTestBase
{
public
class
DistributedCacheTest
extends
StreamingMultipleProgramsTestBase
{
public
static
final
String
data
=
"machen\n"
+
"zeit\n"
...
...
@@ -45,33 +42,31 @@ public class DistributedCacheTest extends JavaProgramTestBase {
+
"keiner\n"
+
"meine\n"
;
protected
String
textPath
;
@Override
protected
void
preSubmit
()
throws
Exception
{
textPath
=
createTempFile
(
"count.txt"
,
data
);
@Test
public
void
testStreamingDistributedCache
()
throws
Exception
{
String
textPath
=
createTempFile
(
"count.txt"
,
data
);
StreamExecutionEnvironment
env
=
StreamExecutionEnvironment
.
getExecutionEnvironment
();
env
.
registerCachedFile
(
textPath
,
"cache_test"
);
env
.
readTextFile
(
textPath
).
flatMap
(
new
WordChecker
());
env
.
execute
();
}
@Override
protected
void
testProgram
()
throws
Exception
{
@Test
public
void
testBatchDistributedCache
()
throws
Exception
{
String
textPath
=
createTempFile
(
"count.txt"
,
data
);
ExecutionEnvironment
env
=
ExecutionEnvironment
.
getExecutionEnvironment
();
env
.
registerCachedFile
(
textPath
,
"cache_test"
);
List
<
Tuple1
<
String
>>
result
=
env
.
readTextFile
(
textPath
)
.
flatMap
(
new
WordChecker
())
.
collect
();
compareResultAsTuples
(
result
,
data
);
env
.
readTextFile
(
textPath
).
flatMap
(
new
WordChecker
()).
count
();
}
public
static
class
WordChecker
extends
RichFlatMapFunction
<
String
,
Tuple1
<
String
>>
{
private
static
final
long
serialVersionUID
=
1L
;
private
final
Set
<
String
>
wordList
=
new
HashSe
t
<>();
private
final
List
<
String
>
wordList
=
new
ArrayLis
t
<>();
@Override
public
void
open
(
Configuration
conf
)
throws
FileNotFoundException
,
IOException
{
public
void
open
(
Configuration
conf
)
throws
IOException
{
File
file
=
getRuntimeContext
().
getDistributedCache
().
getFile
(
"cache_test"
);
BufferedReader
reader
=
new
BufferedReader
(
new
FileReader
(
file
));
String
tempString
;
...
...
@@ -83,9 +78,10 @@ public class DistributedCacheTest extends JavaProgramTestBase {
@Override
public
void
flatMap
(
String
word
,
Collector
<
Tuple1
<
String
>>
out
)
throws
Exception
{
if
(
wordList
.
contains
(
word
))
{
out
.
collect
(
new
Tuple1
<>(
word
));
}
assertTrue
(
"Unexpected word in stream! wordFromStream: "
+
word
+
", shouldBeOneOf: "
+
wordList
.
toString
(),
wordList
.
contains
(
word
));
out
.
collect
(
new
Tuple1
<>(
word
));
}
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录