Flink实践-搭建Flink项目

目录

1. 1. 准备
   1. 1.1. sbt设置
2. 2. Eclipse插件
3. 3. sbt使用生产项目

讲述怎么搭建一个Flink的项目，一开始怎么安装环境。开始Flink开发之旅。

准备

需要下载的文件有sbt，Eclipse，以及插件的网址。

sbt设置

sbt的作用与maven一样，这里不做讲述，可以从sbt官网获取更多信息。下面讲述怎么安装sbt，以及验证sbt安装成功与否。

首先，必须确保开发环境的版本一致，由于开发环境涉及其他开源框架，这里统一采用了sbt-0.13.6版本。
下载sbt-launch.jar,并把它放置在/bin/目录下；然后创建/bin/sbt文件，更改其模式为755，另外填写如下配置到sbt文件里：

SBT_OPTS="-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=256M -Dsbt.override.build.repos=true"
java $SBT_OPTS -jar `dirname $0`/sbt-launch.jar "$@"

然后，是设置库，即sbt构建项目时从何处查找库，一般按照本地优先，然后是公共库，以及特殊库的顺序设置。

在$HOME/.sbt/repositories文件里填入

[repositories]
  local
  my-maven-snapshots:https://repository.apache.org/content/repositories/snapshots/
  my-sbt-releases:https://repository.apache.org/content/repositories/releases/
  my-sbt-public:https://repository.apache.org/content/repositories/public/
  my-maven-public:http://repo.maven.apache.org/maven2/

其他库亦可添加。

最后，验证安装是否成功，在命令行输入sbt –version或者sbt sbt-version，出现sbt版本即成功安装。
另外，sbt下载的包，均在$HOME/.ivy2/cache目录下。

Eclipse插件

开发环境用的是Eclipse，在其中安装scala的插件，打开eclipse点击help下的install new software，然后，填写名字和网站，选择对应的版本，网址信息查找，按照步骤完成安装。

sbt使用生产项目

使用sbt构建项目有多种方法，因为sbt下载依赖的包可以很多方式是设置，下面采取的是一种简单的方式。把各个模块当成是小项目进行生成。

sbt构建项目所需的目录结果和内容，

大目录，项目名称（英文小写）
    build.sbt（是一个文件，用于编辑生成项目所需下载的包）
    project（目录）
       plugins.sbt(eclipse版本设置)
       assembly.sbt（用于打包用）

把上述文件建好后，在对应的文件中添加如下信息，

plugins.sbt

addSbtPlugin("com.typesafe.sbteclipse" % "sbteclipse-plugin" % "2.5.0")

assembly.sbt

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.12.0")

上述两个文件如需添加多个，记得两行之间用一个空行隔开。

最关键是build.sbt文件，是项目生成的最重要的文件，其编辑的格式如下。

lazy val commonSettings = Seq(
  #organization := "****"，
  #version := "0.1.0",
  scalaVersion := "2.11.8",
  #建议使用2.11版本，2.10版本会有很多兼容问题
  libraryDependencies ++= Seq(
    "org.scalatest" %% "scalatest" % "2.2.1" % "test"
  )
)
#版本号根据自身集群环境选择，
val flinkVersion = "****"
val kafkaVersion = "****"
val hbaseVersion = "****"

lazy val flinkSettings = Seq(
  libraryDependencies ++= Seq(
   ("org.apache.flink" % "flink-core" % flinkVersion).
   #为何要排除2.10的，因为在2.11下会自动下载这个2.10的包，导致项目
   #生成失败，所以，必须排除掉，这是多次实践的经验，可以在后续生成结果中查找错误，根据错误进行定位
                 exclude("org.apache.flink","flink-shaded-hadoop1_2.10")
  )
)
#子项目名称
lazy val commonFlink = project.
  settings(commonSettings: _*).
  #包含的配置
  settings(flinkSettings: _*).
  settings(
     libraryDependencies ++= Seq(
        ("org.apache.flink" %% "flink-scala" % flinkVersion).
                  exclude("org.apache.flink","flink-shaded-hadoop1_2.10"),
        ("org.apache.flink" %% "flink-streaming-scala" % flinkVersion).
                  exclude("org.apache.flink","flink-shaded-hadoop1_2.10"),
        ("org.apache.flink" % "flink-streaming-connectors" % flinkVersion).
                  exclude("org.apache.flink","flink-shaded-hadoop1_2.10"),
        ("org.apache.flink" %% "flink-connector-kafka-0.8" % flinkVersion).
                  exclude("org.apache.flink","flink-shaded-hadoop1_2.10"),
        ("org.apache.flink" %% "flink-connector-rabbitmq" % flinkVersion).
                  exclude("org.apache.flink","flink-shaded-hadoop1_2.10"),
        ("org.json4s" %% "json4s-jackson" % "3.2.10"),
        ("org.apache.kafka" %% "kafka" % kafkaVersion)
      )).
  settings(
      #忽略策略，不同开源包存在相同的jar或者class，进行忽略选择，
        assemblyMergeStrategy in assembly <<= (assemblyMergeStrategy in assembly) {
             (old) => {
                   case PathList("META-INF", xs @ _*) => MergeStrategy.discard
                   case x => MergeStrategy.first
             }
         }
     )
#另一个子项目
lazy val example = project.dependsOn(commonFlink).
#每个项目可以依赖其他项目，使用dependsOn来控制，可以多个
  settings(commonSettings: _*).
  settings(flinkSettings: _*).
  settings(
    libraryDependencies ++= Seq(
      "org.apache.hbase" % "hbase-common" % hbaseVersion % "provided",
      "org.apache.hbase" % "hbase-client" % hbaseVersion % "provided",
      "org.scalaj" %% "scalaj-http" % "1.1.4"
    )).
   settings(
        assemblyMergeStrategy in assembly <<= (assemblyMergeStrategy in assembly) {
             (old) => {
                   case PathList("META-INF", xs @ _*) => MergeStrategy.discard
                   case x => MergeStrategy.first
             }
         }
     ).
  settings(runMain in Compile <<= Defaults.runMainTask(fullClasspath in Compile, runner in(Compile, run)))

构建的解析，

1，build.sbt里编辑，具体信息可以查找sbt官网获得更多信息，这里提供几个快捷经验，%%会在对应包自动加上scala的版本，比如加上_2.11或者_2.10；

2，当你不知道某个包的版本号时，可以从maven的开源库中找存在的版本;

3,provided是在打包时不用加载其中，而使用运行集群中的包，或者提交运行时带上的包；为了减少打包的大小，对通用的包可以使用集群的；为了确保程序运行不报错，建议都带上，即去除provided.

4,对于dependsOn的使用，具有传递性，所以不必要把所有的依赖都写上，可以根据传递依赖选择；

5，在sbt生成项目时，用sbt eclipse命令，会很慢是正常的，最好能连vpn，国外网址来加速，请等待，若出现失败，请根据提示定位哪个包下载失败，或者重新sbt eclipse，多次运行；

6，单个子项目中，独立添加库，如下例子，单独添加了下载amidsttoolbox的库源，

.settings(
        resolvers += "my-amisttools" at "http://maven.amidsttoolbox.com/"
).

7,使用sbt可以打包，即sbt projectName/assembly，另在sbt构建项目下，可以设置过滤不要的包，减少打包后的jar大小，具体设置如下：

.settings(
       assemblyExcludedJars in assembly := {
           val cp = (fullClasspath in assembly).value
           cp filter { j =>
              j.data.getName =='****.jar'||
              j.data.getName.contains("***")||
              j.data.getName.startsWith("***")
         }
       }
  )

sbt的项目中使用的一些作用，比如assemblyExcludedJars，assemblyMergeStrategy等，可以从此网站参考；注意的是，如果不知道去除哪些包，可以起初不去除任何包的情况下打包，把记录copy下来，

sbt projectName/assembly > assemblyJar.txt

然后，在文件assemblyJar.txt中查看include的包有哪些，在确认去除哪些，并且parquet前缀的包记得一定打进去。