Flink流计算学习一_大数据系统

Flink流计算学习一

一、flink是什么？
二、使用步骤
- 1.安装hadoop
- 2.配置文件
- 3.创建测试文件
- 4.实现代码
- - 导入依赖
  - 创建文件
  - 实现代码
三、结语

一、flink是什么？

Flink是一个面向数据流处理和批量数据处理的可分布式的开源计算框架，能够支持流处理和批处理两种应用类型。由于流处理和批处理所提供的SLA(服务等级协议)是完全不相同，流处理一般需要支持低延迟、Exactly-once保证（恰好执行一次），而批处理需要支持高吞吐、高效处理，所以在实现的时候通常是分别给出两套实现方法，或者通过一个独立的开源框架来实现其中每一种处理方案。比较典型的有：实现批处理的开源方案有MapReduce、Spark；实现流处理的开源方案有Storm；Spark的Streaming 其实本质上也是微批处理。

二、使用步骤

flink中可以将本地文件，hadoop的hdfs，kafka等作为数据源，在这里我将用hadoop中的hdfs来作为数据源实现。

1.安装hadoop

我是在vmware中一台linux虚拟机来进行试验的，下面来描述具体过程。
首先在虚拟机中创建hadoop文件夹，并且下载并解压hadoop的jar包。

cd /home
mkdir hadoop
cd hadoop
wget http://archive.apache.org/dist/hadoop/common/hadoop-2.8.3/hadoop-2.8.3.tar.gz
tar -xvf hadoop-2.8.3.tar.gz

2.配置文件

配置文件的路径为：/home/hadoop/hadoop-2.8.3/etc/hadoop
core-site.xml


    
        fs.default.name
        hdfs://192.168.1.11:9000

hdfs-site.xml


    
        dfs.replication
        1

mapred-site.xml 解压出来的后面会多出template，删掉即可


    
        mapred.job.tracker
        192.168.1.11:9001

配置已经结束，接下来使用如下命令格式化Hadoop的文件系统HDFS

cd /home/hadoop/hadoop-2.8.3/bin
./hadoop namenode -format

接下来就可以去hadoop下的sbin目录，运行启动命令，将hadoop跑起来。

cd ../sbin
./start-all.sh

运行成功后，就可以通过一下地址访问hadoop与HDFS了。
http://192.168.1.11:8088 (MapReduce的Web页面)
http://192.168.1.11:50070 (HDFS的Web页面)

如果访问不了，就是端口没开放。

同时，vmware虚拟机的网络，最好选择桥接模式，这样重启虚拟机时，IP就不会频繁变动。

3.创建测试文件

计算要有数据源，所以需要去HDFS中创建一个文件，并且开启权限。

cd ../bin
hdfs dfs -touchz /wc.txt
echo "hello word flink oh oh" | ./hdfs dfs -appendToFile - /wc.txt 
./hdfs dfs -chmod -R 777 /

4.实现代码

这里是以scala语言编写的，关于IDEA怎么集成scala自行百度。

导入依赖

	
        
            org.apache.flink
            flink-scala_2.11
            1.9.1
        
        
            org.apache.flink
            flink-streaming-scala_2.11
            1.9.1
        
        
            org.apache.hadoop
            hadoop-common
            2.7.2
        
        
            org.apache.hadoop
            hadoop-client
            2.7.2
        
    
    
         
            
                net.alchim31.maven
                scala-maven-plugin
                3.4.6
                
                     
                        
                            testCompile
                        
                    
                
            
            
                org.apache.maven.plugins
                maven-assembly-plugin
                3.0.0
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single

创建文件

实现代码

package source

import java.net.URL

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object FileSource {

  def main(args: Array[String]): Unit = {
    //初始化上下文
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment;
    //设置并行度(任务可以分散到几个slot运行)
    streamEnv.setParallelism(1);
    //有时代码提示出不来，可以在函数中导入，就会自动提示了
    import org.apache.flink.streaming.api.scala._
    //从hdfs上读取数据源
    val stream: DataStream[String] = streamEnv.readTextFile("hdfs://zjj1:9000/wc.txt")

    val result: DataStream[(String, Int)] = stream
      .flatMap(_.split(" "))//flatMap 读取到的数据按空格分割成为一个数组
      .map((_, 1))//数组中每个元素拆分为一个键值对 key为本身 value为1
      .keyBy(0)//通过key来分组 0是key 1是value
      .sum(1)//用下标为1(即value)的值来计算累加

    result.print();//打印结果

    //执行任务，流计算不执行，不会有结果
    streamEnv.execute("readHdfs");
  }
}

文件中的内容：
通过hdfs dfs -cat /wc.txt来查看文件中的内容

输出结果：

三、结语

今天的学习就到这里了，加油！

Flink流计算学习 一

大数据系统相关栏目本月热门文章

Flink流计算学习一