init project

.gitignore

+*.class
+*.log
+
+# sbt specific
+.cache/
+.history/
+.lib/
+dist/*
+target/
+lib_managed/
+src_managed/
+project/boot/
+project/plugins/project/
+
+# Scala-IDE specific
+.scala_dependencies
+.worksheet
+.idea
+.idea_modules
+.ensime_cache/
+.ensime
+.metals/
+.bloop/
+
+# Play
+.classpath
+.project
+.settings
+RUNNING_PID
+
+# Mill
+out/
\ No newline at end of file

.scalafmt.conf

+version = "2.0.0-RC4"
+maxColumn = 112
+docstrings = JavaDoc
+align = none
+spaces.inImportCurlyBraces = true
+rewrite.rules = [
+  AvoidInfix,
+  RedundantBraces,
+  RedundantParens,
+  PreferCurlyFors
+]

build.sbt

+// ThisBuild / scalaVersion     := "2.12.13"
+ThisBuild / scalaVersion     := "2.11.12"
+ThisBuild / version          := "0.1.0"
+ThisBuild / organization     := "com.gmei"
+
+val esVersion = "7.10.2"
+val es = "org.elasticsearch" % "elasticsearch" % esVersion
+val esCore= "org.elasticsearch" % "elasticsearch-core" % esVersion
+
+val flinkVersion = "1.12.1"
+val flinkCore= "org.apache.flink" % "flink-core" % flinkVersion
+val flinkScala = "org.apache.flink" %% "flink-scala" % flinkVersion
+val flinkStreamingScala = "org.apache.flink" %% "flink-streaming-scala" % flinkVersion
+val flinkJSON = "org.apache.flink" % "flink-json" % flinkVersion % Test
+val flinkRuntime =  "org.apache.flink" %% "flink-runtime" % flinkVersion % Test
+val flinkRuntimeWeb = "org.apache.flink" %% "flink-runtime-web" % flinkVersion % Test
+val flinkClients = "org.apache.flink" %% "flink-clients" % flinkVersion
+val flinkML = "org.apache.flink" % "flink-ml-api" % flinkVersion
+val flinkKafka = "org.apache.flink" %% "flink-connector-kafka" % flinkVersion
+val flinkHadoop = "org.apache.flink" %% "flink-hadoop-compatibility" % flinkVersion % Test
+
+
+lazy val root = (project in file("."))
+  .settings(
+    name := "tryFlink",
+
+    // libraryDependencies += "com.alibaba" % "fastjson" % "1.2.75",
+    libraryDependencies += es,
+    libraryDependencies += esCore,
+
+    libraryDependencies += flinkCore,
+    libraryDependencies += flinkScala,
+    libraryDependencies += flinkStreamingScala,
+    libraryDependencies += flinkClients,
+    libraryDependencies += flinkRuntime,
+    libraryDependencies += flinkRuntimeWeb,
+    // libraryDependencies += flinkJSON,
+    libraryDependencies += flinkML,
+    libraryDependencies += flinkKafka,
+    libraryDependencies += flinkHadoop
+  )
+
+scalacOptions in ThisBuild ++= Seq(
+  "-encoding", "UTF-8",
+  "-deprecation",
+  "-feature",
+  "-unchecked"
+)
+
+mainClass in assembly := Some("Main")
+assemblyJarName in assembly := s"tryFlink.jar"

project/build.properties

+sbt.version=1.3.13
\ No newline at end of file

project/plugins.sbt

+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.15.0")
+addSbtPlugin("ch.epfl.scala" % "sbt-bloop" % "1.4.6-33-1c6f6712")

src/main/scala/com.gmei/Main.scala

+package com.gmei
+
+// import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.scala._
+
+object Main {
+  def main(args: Array[String]): Unit = {
+    println("hello")
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+
+    val socketStream = env.socketTextStream("localhost", 9000)
+
+    // val a = env.fromElements(1, 2, -3, 0, 5, -9, 8)
+    // val b = a.map(_ + 1)
+
+    // b.print()
+
+    val wordsStream = socketStream.flatMap(value => value.split("\\s+")).map(value => (value, 1))
+
+    val keyValuePair = wordsStream.keyBy(0)
+
+    val countPair = keyValuePair.sum(1)
+
+    countPair.print
+
+    env.execute
+  }
+}