flink项目简单搭建在scala环境

r囧r小猫 2021-09-23 07:26 372阅读 0赞

maven文件

<properties>
            <java.version>1.8</java.version>
            <scala.binary.version>2.11</scala.binary.version>
            <flink.version>1.6.1</flink.version>
            <scala.version>2.11.8</scala.version>
            <flinkCore.version>1.8.0</flinkCore.version>
            <clients.version>1.9.0</clients.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_2.11</artifactId>
                <version>1.6.1</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.scala-lang</groupId>
                <artifactId>scala-library</artifactId>
                <version>${scala.version}</version>
                <scope>compile</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flinkCore.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${clients.version}</version>
            </dependency>

小demo代码，默认读取项目根目录下文件

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    
    object Flink {
    
      def main(args:Array[String]) : Unit = {
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    //       下面两种读取数据模式都行
    //    val stream = env.readTextFile("H:/aaaa.txt")
        val stream = env.readTextFile("aaaa.txt")
        stream.print()
    
     //第二种获取数据源
        val stream2 = env.socketTextStream("localhost",11111)
        stream2.print()
    
        env.execute("FirstFlink")
      }
    }

aaaa.txt在项目根目录下

启动时报下面错误

Connected to the target VM, address: '127.0.0.1:50805', transport: 'socket'
    Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/scala/StreamExecutionEnvironment$
    	at com.example.flink.source.Flink$.main(Flink.scala:9)
    	at com.example.flink.source.Flink.main(Flink.scala)
    Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.api.scala.StreamExecutionEnvironment$
    	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    	... 2 more

下载flink包将文件包解压将lib文件下的jar加入项目

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDk1NzUz_size_16_color_FFFFFF_t_70][]

官网下载flink包

![20190911085751370.png][]

** 第二种模式根据ip和端口获取数据**

cmd启动命令窗口 输入 nc -l -p 11111命令，向11111端口输入数据，在cmd中输入数据，启动flink代码stream2获取数据入图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDk1NzUz_size_16_color_FFFFFF_t_70 1][]

上图输出的前面数字3,2,4,1,1，2,3,4是执行打印的线程id，默认8个线程（task）

调用下面语句可将数据写入到text文件下的自定义文本中，会自动生成文本

stream2.writeAsText("H:\\test")

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDk1NzUz_size_16_color_FFFFFF_t_70]: /images/20210923/0c6c92b46b894d5ab5a6e41d55ae1d1b.png
[20190911085751370.png]: /images/20210923/85e089fcccaf4900805cfe66e6eeece9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDk1NzUz_size_16_color_FFFFFF_t_70 1]: /images/20210923/d7ebbaa487a8444ca90d2e0fe7a628e6.png