Flink01-double_happy

介绍

在这里插入图片描述

Stateful Computations over Data Streams
    离线处理/批处理
        state watermark

    事件驱动：
    	Flink ：来一条处理一条
    	Spark：是微批次的  eg：3s     处理 3s的这些数据
    流批一体：

    Exactly-once state consistency
    Event-time processing   ***
    Sophisticated late data handling  延迟数据的处理（延迟数据是避免不了的）
    	（我现在知道的 离线处理 Spark是可以解决延迟数据的问题）

mvn archetype:generate                               \
-DarchetypeGroupId=org.apache.flink              \
-DarchetypeArtifactId=flink-quickstart-scala     \
-DarchetypeVersion=1.9.0



set up the batch execution environment
    ExecutionEnvironment.getExecutionEnvironment
Start with getting some data from the environment
then, transform the resulting DataSet[String] using operations
execute program


(x,...)


RichXXXFunction
生命周期函数
    open   初始化方法
    close  资源释放
    getRuntimeContext  拿到整个作业运行时上下文

Run Applications at any Scale
Flink is designed to run stateful streaming applications at any scale. Applications are parallelized into possibly thousands of tasks that are distributed and concurrently executed in a cluster. Therefore, an application can leverage virtually unlimited amounts of CPUs, main memory, disk and network IO. Moreover, Flink easily maintains very large application state. Its asynchronous and incremental checkpointing algorithm ensures minimal impact on processing latencies while guaranteeing exactly-once state consistency.

Users reported impressive scalability numbers for Flink applications running in their production environments, such as

applications processing multiple trillions of events per day,
applications maintaining multiple terabytes of state, and
applications running on thousands of cores

 maintaining multiple terabytes of state：
 注意：
 	state维护在hdfs 有什么好处和坏处呢？

Flink做的好的地方：
	1.处理时间  eventtime等 
	2.水印
	3.容错机制
	4.状态

pom.xml:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.yt</groupId>
    <artifactId>Flink</artifactId>
    <version>1.0</version>
    <inceptionYear>2008</inceptionYear>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.9.0</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <scala.version>2.11.12</scala.version>
        <hadoop.version>2.7.2</hadoop.version>
    </properties>


    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>


    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>
      <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <!--Flink依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                    <args>
                        <arg>-target:jvm-1.5</arg>
                    </args>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-eclipse-plugin</artifactId>
                <configuration>
                    <downloadSources>true</downloadSources>
                    <buildcommands>
                        <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>
                    </buildcommands>
                    <additionalProjectnatures>
                        <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>
                    </additionalProjectnatures>
                    <classpathContainers>
                        <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
                        <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>
                    </classpathContainers>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <reporting>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                </configuration>
            </plugin>
        </plugins>
    </reporting>
</project>

批处理：
package com.sx.flink01

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.api.scala._
object BatchJob {

  def main(args: Array[String]): Unit = {
    // 获取批处理上下文 <== SparkContext
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 读取数据
    val text: DataSet[String] = env.readTextFile("C:\\IdeaProjects\\flink\\data\\data.txt")

    val value: DataSet[(String, Int)] = text.flatMap(_.toLowerCase.split(","))
      .filter(_.nonEmpty)
      .map((_, 1))

    // transformation
    val result = value.groupBy(0) // 0表示单词
      .sum(1)

    // sink  output
    result.print()
  }
}


结果：
(double_happy,1)
(kite,1)
(25,2)
(kairis,1)
(32,1)

[double_happy@hadoop101 ~]$ nc -lk 7777
doublehappy,1
doublehappy,2
doublehappy,3
-------

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {
    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
  
    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
        val result = text.flatMap(_.toLowerCase.split(","))
          .filter(_.nonEmpty)
          .map((_, 1))
          .keyBy(0).sum(1).print()

    env.execute(this.getClass.getSimpleName)

  }

}

结果是：
2> (2,1)
4> (doublehappy,1)
4> (1,1)
4> (doublehappy,2)
4> (doublehappy,3)
3> (3,1)

[double_happy@hadoop101 ~]$ nc -lk 7777
double,double,double
happy,happy
------------------------

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
        val result = text.flatMap(_.toLowerCase.split(","))
          .filter(_.nonEmpty)
          .map((_, 1))
          .keyBy(0).sum(1).print("double_happy")
          .setParallelism(2)
    env.execute(this.getClass.getSimpleName)

  }
}

结果：
double_happy:1> (double,2)
double_happy:2> (double,1)
double_happy:2> (double,3)
double_happy:1> (happy,1)
double_happy:2> (happy,2)


并行度为1：结果
double_happy> (double,1)
double_happy> (double,2)
double_happy> (double,3)
double_happy> (happy,1)
double_happy> (happy,2)

使用字段表达式

package com.sx.flink01

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamingJob {

  def main(args: Array[String]): Unit = {

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 接收数据
    val text: DataStream[String] = env.socketTextStream("hadoop101", 7777)

    // transformation
//        val result = text.flatMap(_.toLowerCase.split(","))
//          .filter(_.nonEmpty)
//          .map((_, 1))
//          .keyBy(0).sum(1).print("double_happy")
//          .setParallelism(1)

    val result = text.flatMap(_.toLowerCase.split(","))
      .filter(_.nonEmpty)
      .map(x => WC(x, 1))
      .keyBy(_.word).sum("count").print("double_happy")
      .setParallelism(1)
//
    // execute program
    env.execute(this.getClass.getSimpleName)

  }

}

case class WC(word: String, count: Int)

在这里插入图片描述

依赖

在这里插入图片描述

At the bare minimum, the application depends on the Flink APIs. Many applications depend in addition on certain connector libraries (like Kafka, Cassandra, etc.)

注意：
	connector libraries

Specifying Transformation Functions

package com.sx.bean

object Domain {
  case class Access(time:Long, domain:String, traffic:Long)

}

package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })

    accessStream.print()
    env.execute(this.getClass.getSimpleName)
  }
}

结果：
Access(201912120010,ruozedata.com,2000)
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)
Access(201912120010,ruozedata.com,4000)
Access(201912120010,dongqiudi.com,1000)

需求1
	过滤traffic > 4000

方式1：Lambda Functions
object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })
     // 过滤traffic > 4000
     accessStream.filter(_.traffic > 4000).print()
    env.execute(this.getClass.getSimpleName)
  }
}


结果：
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)

方式二：
package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
    val accessStream = stream.map(x => {
      val splits = x.split(",")
      Access(splits(0).toLong, splits(1), splits(2).toLong)
    })
    accessStream.filter(new DoubleHappyFilter02(5000)).print()
    
    //    accessStream.filter(new FilterFunction[Access] {
    //      override def filter(value: Access): Boolean = value.traffic>4000
    //    }).print()     //匿名内部类写法   别这么写  太难看了

    env.execute(this.getClass.getSimpleName)
  }
}

// XXXFunction   RichXXXFunction
class DoubleHappyFilter extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > 4000
}

class DoubleHappyFilter02(traffic: Long) extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > traffic
}


结果：
Access(201912120010,dongqiudi.com,6000)

在这里插入图片描述
这里面的生命周期函数

RichXXXFunction
生命周期函数
    open   初始化方法
    close  资源释放
    getRuntimeContext  拿到整个作业运行时上下文

方式3：Rich functions

package com.sx.flink01

import com.sx.bean.Domain.Access
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object SpecifyingTransformationFunctionsApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    
    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log")
     val accessStream = stream.map(new DoubleHappyMap)

     // 过滤traffic > 4000
     accessStream.filter(_.traffic > 4000).print()
    env.execute(this.getClass.getSimpleName)
  }
}

// XXXFunction   RichXXXFunction
class DoubleHappyFilter extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > 4000
}

class DoubleHappyFilter02(traffic: Long) extends FilterFunction[Access] {
  override def filter(value: Access): Boolean = value.traffic > traffic
}

class DoubleHappyMap extends RichMapFunction[String, Access] {
  override def map(value: String): Access = {
    val splits = value.split(",")
    Access(splits(0).toLong, splits(1), splits(2).toLong)
  }

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    println("~~~~~~~~~~~open~~~~~~~")
  }
  override def close(): Unit = {
    super.close()
  }
  override def getRuntimeContext: RuntimeContext = {
    super.getRuntimeContext
  }
}


结果：  并行度为2的时候  
~~~~~~~~~~~open~~~~~~~
~~~~~~~~~~~open~~~~~~~
1> Access(201912120010,dongqiudi.com,6000)
1> Access(201912120010,zhibo8.com,5000)

这块要注意 ：MySQL获取connection 放在open方法里 
还要注意  不同的并行读  调用了几次 

结果：并行度为1的时候
~~~~~~~~~~~open~~~~~~~
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)