Flink可以在單臺機器上執行，甚至可以在單個Java虛擬機器中執行。 這執行機制可以方便使用者在本地測試和除錯Flink程式。本節概述了Flink的本地執行機制。

本地環境和執行器(executors)允許你可以在本地Java虛擬機器上執行Flink程式，或者是在正在執行程式的Java虛擬機器上(with within any JVM as part of existing programs)。對於大部分示例程式而言，你只需簡單的地點選你IDE上的執行(Run)按鈕就可以執行。

Flink支援兩種不同的本地執行機制: (1) LocalExecutionEnvironment啟動完整的Flink執行環境，包括一個JobManager

1. 除錯

如果你在本地執行Flink程式，還可以像任何其他Java程式一樣來除錯程式。你可以使用System.out.println()來列印一些內部變數，也可以使用偵錯程式。可以在map()，reduce()以及所有其他方法中設定斷點。請參閱Java API文件中的

2. Maven

如果你在Maven專案中開發程式，則必須使用下面依賴關係新增flink-clients模組：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-clients_2.10</artifactId>
  <version>1.3.2</version>
</dependency>

3. 本地執行環境

LocalEnvironment是本地執行Flink程式的控制代碼，可以使用它在本地的JVM，獨立執行或嵌入其他程式裡執行。

本地執行執行環境通過ExecutionEnvironment.createLocalEnvironment()方法例項化。預設情況下，Flink將盡可能使用跟你機器CPU核數一樣多的本地執行緒來執行程式。你可以指定程式你想要的並行度。本地執行環境可以通過enableLogging()/disableLogging()來配置日誌的輸出。

在大多數情況下，ExecutionEnvironment.getExecutionEnvironment()是一種更好的選擇。當程式在本地啟動時(不使用命令列介面)，該方法返回LocalEnvironment，當程式是通過命令列介面提交時，則該方法會返回為在叢集中執行提前配置好的執行環境(pre-configured environment for cluster execution)。

public static void main(String[] args) throws Exception {
    ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();

    DataSet<String> data = env.readTextFile("file:///path/to/file");

    data.filter(new FilterFunction<String>() {
            public boolean filter(String value) {
                return value.startsWith("http://");
            }
        })
        .writeAsText("file:///path/to/result");

    JobExecutionResult res = env.execute();
}

在程式執行結束時會返回JobExecutionResult物件，這個類中包含了程式的執行狀態(runtime)和累加器(accumulator)結果。

LocalEnvironment也可以向Flink傳入使用者自定義配置。

Configuration conf = new Configuration();
conf.setFloat(ConfigConstants.TASK_MANAGER_MEMORY_FRACTION_KEY, 0.5f);
final ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment(conf);

備註:

本地執行環境不啟動任何Web前端來監控執行。

4. 集合執行環境

使用CollectionEnvironment在Java集合上執行，對於執行Flink程式是一種開銷比較低的方法。在這種模式中通常用於自動化測試、除錯、程式碼重用等場景。

使用者可以使用用於批處理的演算法，或者是用於更具互動性的演算法(Users can use algorithms implemented for batch processing also for cases that are more interactive)。Flink程式通過稍微修改就可用於處理請求的Java應用伺服器。

下面是集合環境的例子：

public static void main(String[] args) throws Exception {
    // initialize a new Collection-based execution environment
    final ExecutionEnvironment env = new CollectionEnvironment();

    DataSet<User> users = env.fromCollection( /* get elements from a Java Collection */);

    /* Data Set transformations ... */

    // retrieve the resulting Tuple2 elements into a ArrayList.
    Collection<...> result = new ArrayList<...>();
    resultDataSet.output(new LocalCollectionOutputFormat<...>(result));

    // kick off execution.
    env.execute();

    // Do some work with the resulting ArrayList (=Collection).
    for(... t : result) {
        System.err.println("Result = "+t);
    }
}

flink-examples-batch模組包含一個完整的示例，名稱為CollectionExecutionExample。

備註:

基於集合的Flink程式僅適用於小資料量，這樣可以完全放進JVM堆中。在集合上的執行不是多執行緒的，只使用一個執行緒。

備註:

Flink版本為1.3