Flink实战(五) Batch（DataSet）运算的相关概念

谁借莪１个温暖的怀抱￠ 2023-10-11 10:04 41阅读 0赞

### 文章目录 ###

*   *  1. 示例程序
     *  2. 转换函数
     *  3. 数据源
     *   *  配置CSV解析规则
         *  递归读取输入路径目录
         *  读取压缩文件
     *  4. 数据输出 （Sink）
     *  5. 迭代操作
     *   *   *  Bulk Iterations（\*\*批量迭代\*\*）
             *  Delta Iterations（\*\*增量迭代\*\*）
     *  在函数中操作数据对象
     *   *  Object-Reuse Disabled (DEFAULT)
         *  Object-Reuse Enabled
     *  Debugging
     *   *  Local Execution Environment
         *  Collection Data Sources and Sinks 集合类型的 source 和 Sink
     *  Broadcast Variables 广播变量
     *  Distributed Cache 分布式缓存
     *  Passing Parameters to Functions

DataSet programs in Flink are regular programs that implement transformations on data sets (e.g., filtering, mapping, joining, grouping). The data sets are initially created from certain sources (e.g., by reading files, or from local collections). Results are returned via sinks, which may for example write the data to (distributed) files, or to standard output (for example the command line terminal). Flink programs run in a variety of contexts, standalone, or embedded in other programs. The execution can happen in a local JVM, or on clusters of many machines.

## 1. 示例程序 ##

The following program is a complete, working example of WordCount.

You can copy & paste the code to run it locally.

You only have to include the correct Flink’s library into your project (see Section [Linking with Flink][]) and specify the imports.

Then you are ready to go!

public class WordCountExample { 
        public static void main(String[] args) throws Exception { 
            final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            DataSet<String> text = env.fromElements(
                "Who's there?",
                "I think I hear them. Stand, ho! Who's there?");
    
            DataSet<Tuple2<String, Integer>> wordCounts = text
                .flatMap(new LineSplitter())
                .groupBy(0)
                .sum(1);
    
            wordCounts.print();
        }
    
        public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { 
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) { 
                for (String word : line.split(" ")) { 
                    out.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        }
    }

## 2. 转换函数 ##

Data transformations transform one or more DataSets into a new DataSet. Programs can combine multiple transformations into sophisticated assemblies.

转换函数可以转换一个或者多个DataSet为一个新的Dataset，程序可以将多个转换组合成复杂的程序集。

## 3. 数据源 ##

File-based: 文件类型

*  `readTextFile(path)` / `TextInputFormat` \- Reads files line wise and returns them as Strings.
 *  `readTextFileWithValue(path)` / `TextValueInputFormat` \- Reads files line wise and returns them as StringValues. StringValues are mutable strings.
 *  `readCsvFile(path)` / `CsvInputFormat` \- Parses files of comma (or another char) delimited fields. Returns a DataSet of tuples or POJOs. Supports the basic java types and their Value counterparts as field types.
 *  `readFileOfPrimitives(path, Class)` / `PrimitiveInputFormat` \- Parses files of new-line (or another char sequence) delimited primitive data types such as `String` or `Integer`.
 *  `readFileOfPrimitives(path, delimiter, Class)` / `PrimitiveInputFormat` \- Parses files of new-line (or another char sequence) delimited primitive data types such as `String` or `Integer` using the given delimiter.

Collection-based: 集合类型

*  `fromCollection(Collection)` \- Creates a data set from a Java.util.Collection. All elements in the collection must be of the same type.
 *  `fromCollection(Iterator, Class)` \- Creates a data set from an iterator. The class specifies the data type of the elements returned by the iterator.
 *  `fromElements(T ...)` \- Creates a data set from the given sequence of objects. All objects must be of the same type.
 *  `fromParallelCollection(SplittableIterator, Class)` \- Creates a data set from an iterator, in parallel. The class specifies the data type of the elements returned by the iterator.
 *  `generateSequence(from, to)` \- Generates the sequence of numbers in the given interval, in parallel.

Generic: 通用的

*  `readFile(inputFormat, path)` / `FileInputFormat` \- Accepts a file input format.
 *  `createInput(inputFormat)` / `InputFormat` \- Accepts a generic input format.

--------------------

### 配置CSV解析规则 ###

Flink offers a number of configuration options for CSV parsing:

*  `types(Class ... types)` specifies the types of the fields to parse. **It is mandatory to configure the types of the parsed fields.** In case of the type class Boolean.class, “True” (case-insensitive), “False” (case-insensitive), “1” and “0” are treated as booleans.
 *  `lineDelimiter(String del)` specifies the delimiter of individual records. The default line delimiter is the new-line character `'\n'`.
 *  `fieldDelimiter(String del)` specifies the delimiter that separates fields of a record. The default field delimiter is the comma character `','`.
 *  `includeFields(boolean ... flag)`, `includeFields(String mask)`, or `includeFields(long bitMask)` defines which fields to read from the input file (and which to ignore). By default the first *n* fields (as defined by the number of types in the `types()` call) are parsed.
 *  `parseQuotedStrings(char quoteChar)` enables quoted string parsing. Strings are parsed as quoted strings if the first character of the string field is the quote character (leading or tailing whitespaces are *not* trimmed). Field delimiters within quoted strings are ignored. Quoted string parsing fails if the last character of a quoted string field is not the quote character or if the quote character appears at some point which is not the start or the end of the quoted string field (unless the quote character is escaped using ‘’). If quoted string parsing is enabled and the first character of the field is *not* the quoting string, the string is parsed as unquoted string. By default, quoted string parsing is disabled.
 *  `ignoreComments(String commentPrefix)` specifies a comment prefix. All lines that start with the specified comment prefix are not parsed and ignored. By default, no lines are ignored.
 *  `ignoreInvalidLines()` enables lenient parsing, i.e., lines that cannot be correctly parsed are ignored. By default, lenient parsing is disabled and invalid lines raise an exception.
 *  `ignoreFirstLine()` configures the InputFormat to ignore the first line of the input file. By default no line is ignored.

### 递归读取输入路径目录 ###

For file-based inputs, when the input path is a directory, nested files are not enumerated by default. Instead, only the files inside the base directory are read, while nested files are ignored. Recursive enumeration of nested files can be enabled through the `recursive.file.enumeration` configuration parameter, like in the following example.

// enable recursive enumeration of nested input files
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
    // create a configuration object
    Configuration parameters = new Configuration();
    
    // set the recursive enumeration parameter
    parameters.setBoolean("recursive.file.enumeration", true);
    
    // pass the configuration to the data source
    DataSet<String> logs = env.readTextFile("file:///path/with.nested/files")
    			  .withParameters(parameters);

### 读取压缩文件 ###

Flink currently supports transparent decompression of input files if these are marked with an appropriate file extension. In particular, this means that no further configuration of the input formats is necessary and any `FileInputFormat` support the compression, including custom input formats. Please notice that compressed files might not be read in parallel, thus impacting job scalability.

The following table lists the currently supported compression methods.

<table> 
 <thead> 
  <tr> 
   <th align="left">Compression method</th> 
   <th align="left">File extensions</th> 
   <th align="left">Parallelizable</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong>DEFLATE</strong></td> 
   <td align="left"><code>.deflate</code></td> 
   <td align="left">no</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>GZip</strong></td> 
   <td align="left"><code>.gz</code>, <code>.gzip</code></td> 
   <td align="left">no</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Bzip2</strong></td> 
   <td align="left"><code>.bz2</code></td> 
   <td align="left">no</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>XZ</strong></td> 
   <td align="left"><code>.xz</code></td> 
   <td align="left">no</td> 
  </tr> 
 </tbody> 
</table>

## 4. 数据输出 （Sink） ##

Data sinks consume DataSets and are used to store or return them. Data sink operations are described using an [OutputFormat][]. Flink comes with a variety of built-in output formats that are encapsulated behind operations on the DataSet:

*  `writeAsText()` / `TextOutputFormat` \- Writes elements line-wise as Strings. The Strings are obtained by calling the *toString()* method of each element.
 *  `writeAsFormattedText()` / `TextOutputFormat` \- Write elements line-wise as Strings. The Strings are obtained by calling a user-defined *format()* method for each element.
 *  `writeAsCsv(...)` / `CsvOutputFormat` \- Writes tuples as comma-separated value files. Row and field delimiters are configurable. The value for each field comes from the *toString()* method of the objects.
 *  `print()` / `printToErr()` / `print(String msg)` / `printToErr(String msg)` \- Prints the *toString()* value of each element on the standard out / standard error stream. Optionally, a prefix (msg) can be provided which is prepended to the output. This can help to distinguish between different calls to *print*. If the parallelism is greater than 1, the output will also be prepended with the identifier of the task which produced the output.
 *  `write()` / `FileOutputFormat` \- Method and base class for custom file outputs. Supports custom object-to-bytes conversion.
 *  `output()`/ `OutputFormat` \- Most generic output method, for data sinks that are not file based (such as storing the result in a database).

## 5. 迭代操作 ##

Iterations implement loops in Flink programs.

在Flink程序中通过迭代实现循环。

The iteration operators encapsulate a part of the program and execute it repeatedly, feeding back the result of one iteration (the partial solution) into the next iteration.

迭代运算符封装程序的一部分并把它重复执行，将一次迭代的结果（部分结果）反馈到下一次迭代中。

There are two types of iterations in Flink: **BulkIteration** and **DeltaIteration**.

Flink中有两种类型的迭代 **BulkIteration** 和**DeltaIteration**.

#### Bulk Iterations（**批量迭代**） ####

To create a BulkIteration call the `iterate(int)` method of the DataSet the iteration should start at.

要创建BulkIteration，请调用迭代的DataSet的iterate（int）方法。

This will return an `IterativeDataSet`, which can be transformed with the regular operators.

这将返回一个IterativeDataSet，它可以使用常规运算符进行转换

The single argument to the iterate call specifies the maximum number of iterations.

迭代调用的单个参数指定最大迭代次数。

To specify the end of an iteration call the `closeWith(DataSet)` method on the `IterativeDataSet` to specify which transformation should be fed back to the next iteration.

要指定迭代的结束，请在IterativeDataSet上调用closeWith（DataSet）方法，以指定应将哪个转换反馈到下一次迭代。

You can optionally specify a termination criterion with `closeWith(DataSet, DataSet)`, which evaluates the second DataSet and terminates the iteration, if this DataSet is empty.

您可以选择使用closeWith（DataSet，DataSet）指定终止条件，如果此DataSet为空，则会评估第二个DataSet并终止迭代。

If no termination criterion is specified, the iteration terminates after the given maximum number iterations.

如果未指定终止条件，则迭代将在给定的最大数量迭代后终止。

The following example iteratively estimates the number Pi.

以下示例迭代地估计数量Pi。

The goal is to count the number of random points, which fall into the unit circle.

目标是计算落入单位圆的随机点数。

In each iteration, a random point is picked. If this point lies inside the unit circle, we increment the count. Pi is then estimated as the resulting count divided by the number of iterations multiplied by

在每次迭代中，挑选一个随机点。如果此点位于单位圆内，我们会增加计数。然后估计Pi作为结果计数除以迭代次数乘以4。

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
    // Create initial IterativeDataSet
    IterativeDataSet<Integer> initial = env.fromElements(0).iterate(10000);
    
    DataSet<Integer> iteration = initial.map(new MapFunction<Integer, Integer>() {
        @Override
        public Integer map(Integer i) throws Exception {
            double x = Math.random();
            double y = Math.random();
    
            return i + ((x * x + y * y < 1) ? 1 : 0);
        }
    });
    
    // Iteratively transform the IterativeDataSet
    DataSet<Integer> count = initial.closeWith(iteration);
    
    count.map(new MapFunction<Integer, Double>() {
        @Override
        public Double map(Integer count) throws Exception {
            return count / (double) 10000 * 4;
        }
    }).print();
    
    env.execute("Iterative Pi Example");

You can also check out the [K-Means example][], which uses a BulkIteration to cluster a set of unlabeled points.

您还可以查看K-Means示例，该示例使用BulkIteration对一组未标记的点进行聚类。

#### Delta Iterations（**增量迭代**） ####

Delta iterations exploit the fact that certain algorithms do not change every data point of the solution in each iteration.

Delta迭代利用了某些算法在每次迭代中不改变结果的每个数据点的事实。

In addition to the partial solution that is fed back (called workset) in every iteration, delta iterations maintain state across iterations (called solution set), which can be updated through deltas.

除了在每次迭代中反馈的部分解决方案（称为工作集）之外，增量迭代还保持了跨越迭代维护的状态(被叫做解集)，，可以通过增量更新。

The result of the iterative computation is the state after the last iteration. Please refer to the [Introduction to Iterations][] for an overview of the basic principle of delta iterations.

迭代计算的结果是上一次迭代之后的状态。有关delta迭代的基本原理的概述，请参阅迭代简介。

Defining a DeltaIteration is similar to defining a BulkIteration. For delta iterations, two data sets form the input to each iteration (workset and solution set), and two data sets are produced as the result (new workset, solution set delta) in each iteration.

定义DeltaIteration类似于定义BulkIteration。对于delta迭代，两个数据集形成每次迭代的输入（工作集和答案集），并且在每次迭代中生成两个数据集作为结果（新工作集，增量解集）。

To create a DeltaIteration call the `iterateDelta(DataSet, int, int)` (or `iterateDelta(DataSet, int, int[])` respectively). This method is called on the initial solution set. The arguments are the initial delta set, the maximum number of iterations and the key positions. The returned `DeltaIteration` object gives you access to the DataSets representing the workset and solution set via the methods `iteration.getWorkset()` and `iteration.getSolutionSet()`.

要创建DeltaIteration，请分别调用iterateDelta（DataSet，int，int）（或iterateDelta（DataSet，int，int \[\]））在初始解决方案集上调用此方法。参数是初始增量集，最大迭代次数和关键位置。返回的DeltaIteration对象使您可以通过方法iteration.getWorkset（）和iteration.getSolutionSet（）访问表示工作集和答案集的DataSet。

Below is an example for the syntax of a delta iteration

// read the initial data sets
    DataSet<Tuple2<Long, Double>> initialSolutionSet = // [...]
    
    DataSet<Tuple2<Long, Double>> initialDeltaSet = // [...]
    
    int maxIterations = 100;
    int keyPosition = 0;
    
    DeltaIteration<Tuple2<Long, Double>, Tuple2<Long, Double>> iteration = initialSolutionSet
        .iterateDelta(initialDeltaSet, maxIterations, keyPosition);
    
    DataSet<Tuple2<Long, Double>> candidateUpdates = iteration.getWorkset()
        .groupBy(1)
        .reduceGroup(new ComputeCandidateChanges());
    
    DataSet<Tuple2<Long, Double>> deltas = candidateUpdates
        .join(iteration.getSolutionSet())
        .where(0)
        .equalTo(0)
        .with(new CompareChangesToCurrent());
    
    DataSet<Tuple2<Long, Double>> nextWorkset = deltas
        .filter(new FilterByThreshold());
    
    iteration.closeWith(deltas, nextWorkset)
    	.writeAsCsv(outputPath);

## 在函数中操作数据对象 ##

Operating on data objects in functions

Flink’s runtime exchanges data with user functions in form of Java objects. Functions receive input objects from the runtime as method parameters and return output objects as result. Because these objects are accessed by user functions and runtime code, it is very important to understand and follow the rules about how the user code may access, i.e., read and modify, these objects.

Flink的运行时以Java对象的形式与用户函数交换数据。 函数从运行时接收输入对象作为方法参数，并返回输出对象作为结果。 因为这些对象是由用户函数和运行时代码访问的，所以理解并遵循有关用户代码如何访问（即读取和修改）这些对象的规则非常重要。

User functions receive objects from Flink’s runtime either as regular method parameters (like a `MapFunction`) or through an `Iterable` parameter (like a `GroupReduceFunction`). We refer to objects that the runtime passes to a user function as *input objects*. User functions can emit objects to the Flink runtime either as a method return value (like a `MapFunction`) or through a `Collector` (like a `FlatMapFunction`). We refer to objects which have been emitted by the user function to the runtime as *output objects*.

用户函数从Flink的运行时接收对象，作为常规方法参数（如MapFunction）或通过Iterable参数（如GroupReduceFunction）。 我们将运行时传递给用户函数的对象称为输入对象。 用户函数可以将对象作为方法返回值（如MapFunction）或通过收集器（如FlatMapFunction）发送到Flink运行时。 我们将用户函数发出的对象称为输出对象。

Flink’s DataSet API features two modes that differ in how Flink’s runtime creates or reuses input objects. This behavior affects the guarantees and constraints for how user functions may interact with input and output objects. The following sections define these rules and give coding guidelines to write safe user function code.

Flink的DataSet API具有两种模式，这些模式在Flink的运行时创建或重用输入对象方面有所不同。 此行为会影响用户函数如何与输入和输出对象进行交互的保证和约束。 以下部分定义了这些规则，并给出了编写安全用户功能代码的编码指南。

### Object-Reuse Disabled (DEFAULT) ###

By default, Flink operates in object-reuse disabled mode. This mode ensures that functions always receive new input objects within a function call. The object-reuse disabled mode gives better guarantees and is safer to use. However, it comes with a certain processing overhead and might cause higher Java garbage collection activity. The following table explains how user functions may access input and output objects in object-reuse disabled mode.

默认情况下，Flink禁用了是对象重用模式，这个模式保证了函数每一次获得的输入对象都是最新的。禁用对象重用模式可提供更好的保证，并且使用起来更安全。但是，它带来了一定的处理开销，可能会导致更高的Java垃圾回收活动。 下表说明了用户函数如何在禁用对象重用模式下访问输入和输出对象。

<table> 
 <thead> 
  <tr> 
   <th align="left">Operation</th> 
   <th align="center">Guarantees and Restrictions</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong>Reading Input Objects</strong></td> 
   <td align="center">Within a method call it is guaranteed that the value of an input object does not change. This includes objects served by an Iterable. For example it is safe to collect input objects served by an Iterable in a List or Map. Note that objects may be modified after the method call is left. It is <strong>not safe</strong> to remember objects across function calls.<br>在方法调用中，保证输入对象的值不会改变。 这包括由Iterable提供的对象。 例如，收集由List或Map中的Iterable提供的输入对象是安全的。 请注意，在离开方法调用后，可以修改对象。 记住函数调用中的对象是不安全的**。</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Modifying Input Objects</strong></td> 
   <td align="center">You may modify input objects.<br>您可以修改输入对象。</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Emitting Input Objects</strong></td> 
   <td align="center">You may emit input objects. The value of an input object may have changed after it was emitted. It is <strong>not safe</strong> to read an input object after it was emitted.<br>您可以发出输入对象。 输入对象的值在发出后可能已更改。 在输出对象之后读取输入对象是不安全的。</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Reading Output Objects</strong></td> 
   <td align="center">An object that was given to a Collector or returned as method result might have changed its value. It is <strong>not safe</strong> to read an output object.</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Modifying Output Objects</strong></td> 
   <td align="center">You may modify an object after it was emitted and emit it again.</td> 
  </tr> 
 </tbody> 
</table>

**Coding guidelines for the object-reuse disabled (default) mode:**

*  Do not remember and read input objects across method calls. 不记得并跨方法调用读取输入对象。
 *  Do not read objects after you emitted them. 发射后不要读取对象。

### Object-Reuse Enabled ###

In object-reuse enabled mode, Flink’s runtime minimizes the number of object instantiations. This can improve the performance and can reduce the Java garbage collection pressure. The object-reuse enabled mode is activated by calling `ExecutionConfig.enableObjectReuse()`. The following table explains how user functions may access input and output objects in object-reuse enabled mode.

在对象重用启用模式下，Flink的运行时最小化对象实例化的数量。 这可以提高性能并且可以减少Java垃圾收集压力。 通过调用`ExecutionConfig.enableObjectReuse（）`激活对象重用启用模式。 下表说明了用户函数如何在对象重用启用模式下访问输入和输出对象。

<table> 
 <thead> 
  <tr> 
   <th align="left">Operation</th> 
   <th align="center">Guarantees and Restrictions(保证和限制)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><strong>Reading input objects received as regular method parameters</strong></td> 
   <td align="center">Input objects received as regular method arguments are not modified within a function call. Objects may be modified after method call is left. It is <strong>not safe</strong> to remember objects across function calls.</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Reading input objects received from an Iterable parameter</strong></td> 
   <td align="center">Input objects received from an Iterable are only valid until the next() method is called. An Iterable or Iterator may serve the same object instance multiple times. It is <strong>not safe</strong> to remember input objects received from an Iterable, e.g., by putting them in a List or Map.</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Modifying Input Objects</strong></td> 
   <td align="center">You <strong>must not</strong> modify input objects, except for input objects of MapFunction, FlatMapFunction, MapPartitionFunction, GroupReduceFunction, GroupCombineFunction, CoGroupFunction, and InputFormat.next(reuse).</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Emitting Input Objects</strong></td> 
   <td align="center">You <strong>must not</strong> emit input objects, except for input objects of MapFunction, FlatMapFunction, MapPartitionFunction, GroupReduceFunction, GroupCombineFunction, CoGroupFunction, and InputFormat.next(reuse).</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Reading Output Objects</strong></td> 
   <td align="center">An object that was given to a Collector or returned as method result might have changed its value. It is <strong>not safe</strong> to read an output object.</td> 
  </tr> 
  <tr> 
   <td align="left"><strong>Modifying Output Objects</strong></td> 
   <td align="center">You may modify an output object and emit it again.</td> 
  </tr> 
 </tbody> 
</table>

**Coding guidelines for object-reuse enabled:**

*  Do not remember input objects received from an `Iterable`.
 *  Do not remember and read input objects across method calls.
 *  Do not modify or emit input objects, except for input objects of `MapFunction`, `FlatMapFunction`, `MapPartitionFunction`, `GroupReduceFunction`, `GroupCombineFunction`, `CoGroupFunction`, and `InputFormat.next(reuse)`.
 *  To reduce object instantiations, you can always emit a dedicated(专用) output object which is repeatedly modified but never read.

## Debugging ##

Before running a data analysis program on a large data set in a distributed cluster, it is a good idea to make sure that the implemented algorithm works as desired. Hence, implementing data analysis programs is usually an incremental process of checking results, debugging, and improving.

Flink provides a few nice features to significantly ease the development process of data analysis programs by supporting local debugging from within an IDE, injection of test data, and collection of result data. This section give some hints how to ease the development of Flink programs.

Flink提供了一些很好的功能，通过支持IDE内的本地调试，测试数据的注入和结果数据的收集，显着简化了数据分析程序的开发过程。 本节提供了一些如何简化Flink程序开发的提示。

### Local Execution Environment ###

A `LocalEnvironment` starts a Flink system within the same JVM process it was created in. If you start the LocalEnvironment from an IDE, you can set breakpoints in your code and easily debug your program.

A LocalEnvironment is created and used as follows:

final ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
    
    DataSet<String> lines = env.readTextFile(pathToTextFile);
    // build your program
    
    env.execute();

### Collection Data Sources and Sinks 集合类型的 source 和 Sink ###

Providing input for an analysis program and checking its output is cumbersome when done by creating input files and reading output files. Flink features special data sources and sinks which are backed by Java collections to ease testing. Once a program has been tested, the sources and sinks can be easily replaced by sources and sinks that read from / write to external data stores such as HDFS.

## Broadcast Variables 广播变量 ##

Broadcast variables allow you to make a data set available to all parallel instances of an operation, in addition to the regular input of the operation. This is useful for auxiliary data sets, or data-dependent parameterization. The data set will then be accessible at the operator as a Collection.

除了操作的常规输入之外，广播变量还允许使数据集可用于操作的所有并行实例。 这对于辅助数据集或与数据相关的参数设置很有用。 然后可以将数据集作为集合进行访问。

*  **Broadcast**: broadcast sets are registered by name via `withBroadcastSet(DataSet, String)`, and
 *  **Access**: accessible via `getRuntimeContext().getBroadcastVariable(String)` at the target operator.

## Distributed Cache 分布式缓存 ##

Flink offers a distributed cache, similar to Apache Hadoop, to make files locally accessible to parallel instances of user functions. This functionality can be used to share files that contain static external data such as dictionaries or machine-learned regression models.

Flink提供类似于Apache Hadoop的分布式缓存，以使文件在本地可被并行执行的实例中的函数访问。 此功能可用于共享包含静态外部数据的文件，例如字典或机器学习的回归模型。

The cache works as follows. A program registers a file or directory of a [local or remote filesystem such as HDFS or S3][] under a specific name in its `ExecutionEnvironment` as a cached file. When the program is executed, Flink automatically copies the file or directory to the local filesystem of all workers. A user function can look up the file or directory under the specified name and access it from the worker’s local filesystem.

缓存的工作方式如下。

1.  程序注册\[本地或远程文件系统，例如HDFS或S3\]的文件或目录（https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/batch/connectors.html ＃reading-from-file-systems）在其`ExecutionEnvironment`中以特定名称作为缓存文件。
2.  程序执行后，`Flink`会自动将文件或目录复制到所有`Worker`的本地文件系统中。`Worker`的编写的程序可以查找指定名称下的文件或目录，并从`Worker`的本地文件系统中访问它。

## Passing Parameters to Functions ##

Parameters can be passed to functions using either the constructor or the `withParameters(Configuration)` method. The parameters are serialized as part of the function object and shipped to all parallel task instances.

可以使用构造函数或`withParameters（Configuration）`方法将参数传递给函数。 参数被序列化为功能对象的一部分，并被运送到所有并行任务实例。

[Linking with Flink]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/projectsetup/dependencies.html
[OutputFormat]: https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java
[K-Means example]: https://github.com/apache/flink/blob/master//flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java
[Introduction to Iterations]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/batch/iterations.html
[local or remote filesystem such as HDFS or S3]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/batch/connectors.html#reading-from-file-systems