Flink開發-IDEA scala開發環境搭建

現在大數據相關服務，越來越傾向於使用scala語言，scala函數式編程的優勢我不多贅述。最明顯的一個優點，代碼簡潔。看個WordCount實現對比：

Java版WordCount

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

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

DataSet<Tuple2<String, Integer>> counts =
        // split up the lines in pairs (2-tuples) containing: (word,1)
        text.flatMap(new Tokenizer())
        // group by the tuple field "0" and sum up tuple field "1"
        .groupBy(0)
        .sum(1);

counts.writeAsCsv(outputPath, "\n", " ");

// User-defined functions
public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {

    @Override
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        // normalize and split the line
        String[] tokens = value.toLowerCase().split("\\W+");

        // emit the pairs
        for (String token : tokens) {
            if (token.length() > 0) {
                out.collect(new Tuple2<String, Integer>(token, 1));
            }   
        }
    }
}

 

scala版WordCount

val env = ExecutionEnvironment.getExecutionEnvironment

// get input data
val text = env.readTextFile("/path/to/file")

val counts = text.flatMap { _.toLowerCase.split("\\s+") filter { _.nonEmpty } }
  .map ( (_, 1) )
  .groupBy(0)
  .sum(1)

counts.writeAsCsv(outputPath, "\n", " ")

 

如何搭建開發環境呢？

這里介紹一下主要關注的點。

• 版本匹配

idea和scala的版本匹配有比較大的耦合，具體如何對應需要確認一下。根據flink的scala版本選定要安裝的scala sdk，flink使用的scala sdk是2.12，所以我這里選擇的sdk版本是2.12.8。

• 安裝scala plugin

 

• 設置Library

創建flink maven工程后，右鍵工程打開module setting，在Libraries中下載所需版本的scala sdk，選擇下載並等待下載成功。

• 避免多版本sdk沖突

確認安裝成功，如果存在其他版本的scala sdk，刪除掉

 

Global Libraries

Global Libraries中，同樣選中所需的scala sdk，否則compile可能出錯，類似這樣的錯誤 compiler error

• maven依賴

使用flink scala版本的包依賴，官方的例子用的java版本依賴，會導致寫scala調用一些flink api出現語法錯誤。

 

<version>1.0-SNAPSHOT</version>
<dependencies>
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.12</artifactId>
        <version>1.9.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-scala -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-scala_2.12</artifactId>
        <version>1.9.0</version>
    </dependency>
</dependencies>

 

這樣代碼可以正常在IDEA里面跑了。

 

 

如果創建時找不到scala的文件

 

需要導入scala的sdk，忘了說了，別忘了裝好 1.8的java的sdk

 

 

 

import org.apache.flink.api.scala.ExecutionEnvironment
object WordCountBatch {
  def main(args: Array[String]): Unit = {
 
    val inputPath = "D:\\data\\11.txt"
//    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val env = ExecutionEnvironment.getExecutionEnvironment
 
    import  org.apache.flink.api.scala._
    val text = env.readTextFile(inputPath)
    val counts = text.flatMap(_.split("\\W+"))
      .filter(_.nonEmpty)
      .map((_,1))
      .groupBy(0)
      .sum(1)
 
    counts.writeAsCsv("D:\\data\\output6").setParallelism(1)
    env.execute("batch wordCount")
  }
 
}