美文网首页
learning flink

learning flink

作者: 奉先 | 来源:发表于2019-11-14 14:58 被阅读0次

    2. 环境准备

    (2019-11-12)

    2.1 项目模板

    Flink执行环境分为本地环境和集群环境,一般地,为了正确执行flink需要JDK环境/Scala环境,Maven环境,Hadoop环境。
    为了快速搭建,针对java和scala官网提供了项目模板来快速创建项目,官网参考地址:
    java版本项目模板
    scala版本项目模板
    官网提供2种方式快速构建项目,maven和脚本方式(需要自行制定版本):

        $ mvn archetype:generate                               \
          -DarchetypeGroupId=org.apache.flink              \
          -DarchetypeArtifactId=flink-quickstart-scala     \
          -DarchetypeVersion=1.9.0
    
        $ curl https://flink.apache.org/q/quickstart-scala.sh | bash -s 1.9.0
    

    2.2 Flink开发环境

    1. scala插件安装:
      官方建议使用IDEA,所以需要安装。另外,需要安装Scala插件,具体方法如下:
      Preferences -> Plugins
      如果未安装过scala插件, 在MarketPlace 搜索scala,Install即可。



      安装后重启IDE,完成插件的安装。

    2.导入项目
    项目创建后,使用IDEA打开,可以看到项目结构:



    Java的默认JVM堆大小对于Flink可能太小,建议手工增加。

    2.3 配置依赖

    关于Maven依赖的scope设置的备用知识:maven依赖的scope含义
    依赖配置项:

            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            <!-- Scala Library, provided by Flink as well. -->
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>${scala.version}</version>
                <scope>provided</scope>
            </dependency>
    

    建议将依赖的scope项配置为“provided”,否则会使生成的JAR变得过大,因为它还包含所有Flink核心依赖项。更坏可能是,添加到应用程序jar文件中的Flink核心依赖项与您自己的某些依赖项版本发生冲突。

    大多数flink应用程序都需要特定的连接器或库来运行,例如与Kafka,Cassandra等的连接器。添加如下依赖,并且打包时,一同打包上传。

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
        <version> 1.9.0 </version>
    </dependency>
    

    2.4 Scala REPL

    官网参考:Scala REPL
    为了实验此节,需要先安装一个本地,下载scala的安装包,直接解压:

    $ tar zxvf flink-1.9.1-bin-scala_2.11.tgz
    $ cd flink-1.9.1/
    $ export TERM=xterm-color
    $ bin/start-scala-shell.sh local       #启动scala REPL 本地
    

    Flink自带一个集成的scala shell交互,即Scala REPL。Scala REPL支持DataSet,DataStream,Table API和SQL。
    上边启动Scala REPL遇到了几个坑:1.先不用本地启动集群,直接启动Scala REPL,否则的话报错。 2. 直接启动时,报了一个exception:java.lang.NumberFormatException: For input string: "0x100",想解决需要执行下$ export TERM=xterm-color,(由于终端一些样式格式的问题导致)。
    当启动Scala REPL时,Flink已经初始化好了相应的Environment,分别使用“benv”和“senv”变量获取批量和流式计算环境。
    下面是官网的例子来实现批量和流式的WordCount:

    scala> val text = benv.fromElements(
         |   "To be, or not to be,--that is the question:--",
         |   "Whether 'tis nobler in the mind to suffer",
         |   "The slings and arrows of outrageous fortune",
         |   "Or to take arms against a sea of troubles,")
    text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@1f3f0d25
    scala> val counts = text.flatMap{ _.toLowerCase.split("\\W+") }.map { (_, 1) }.groupBy(0).sum(1)
    counts: org.apache.flink.api.scala.AggregateDataSet[(String, Int)] = org.apache.flink.api.scala.AggregateDataSet@83b0d9f
    
    scala> counts.print()
    

    1.在scala REPL交互式窗口,想退出的话,输入命令 :q

    2.5 Flink源码编译

    下载源码包,既可以从github上clone下来,也可以在官网download页直接下载源代码,源代码包很小。

    $ git clone https://github.com/apache/flink
    

    3. Flink编程模型

    (2019-11-13)
    有界数据集具有时间边界,无界数据集没有时间边界;对有界数据的处理称为批计算,对无界数据的处理称为流计算;支持批处理的API称为DataSet API,支持流处理的API称为DataStream API。

    相关文章

      网友评论

          本文标题:learning flink

          本文链接:https://www.haomeiwen.com/subject/odqrictx.html