Flink03--double_happy

写东西 还是要有通用的思想

通用的！！！ 与底层的执行引擎是没关系的  Beam
    Spark
        1.6
        2.x
    Flink

    你的API上一定不能出现底层执行引擎的API

    适配 ==> Spark/Flink

上一篇文章 ：
MySQLSource：
class MySQLSource extends RichSourceFunction[Student]{

  var connection:Connection = _
  var pstmt:PreparedStatement = _

  // 在open方法中建立连接
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    connection = MySQLUtils.getConnection("hadoop101","3306","test","root", "wsx123$%^")
    pstmt = connection.prepareStatement("select * from student")
  }

  // 释放
  override def close(): Unit = {
    super.close()
    MySQLUtils.closeConnection(connection, pstmt)
  }

  override def cancel(): Unit = {

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {

    println("~~~~run~~~~~~")
    val rs = pstmt.executeQuery()
    while(rs.next()){
      val student = Student(rs.getInt("id"), rs.getString("name"), rs.getString("password"),rs.getInt("age"))
      ctx.collect(student)
    }
  }
}

注意：
class MySQLSource extends RichSourceFunction[Student]

RichSourceFunction：
这个东西 并不是 上一篇文章的三个Function 中的一个 

那么 RichSourceFunction 的并行度是多少？


源码：
public abstract class RichSourceFunction<OUT> extends AbstractRichFunction implements SourceFunction<OUT> {

	private static final long serialVersionUID = 1L;
}

我当时测试的时候 设置并行度为3的时候 报错的 

源码里是继承SourceFunction的 
所以并行度是1

那么就有一个问题？
 有两个流：
 
	stream1  4并行度  emp表  log里      domain
	stream2  1并行度  dept表  MySQL里    user_id
这两个流进行 
stream1.connect(stream2).map(x=>{})

即：
	logStream emp: deptno empno ename
	mysqlStream dept: deptno dname
	
	outputStream empno,ename,deptno,dname
	
	数据清洗/ETL
	outputStream = stream1.connect(stream2).map(x=>{
	    ....
	})

那么connect会成功么？？
	4并行度的流 里面有写task 的deptno 关联 
	1并行度的流 拿取dname 是拿不到的 
那么这个问题该怎么解决呢？之后再说

先解决并行度为1的问题 下面
 变为：RichParallelSourceFunction

class ScalikeJDBCMySQLSource extends RichParallelSourceFunction[Student]{

  override def cancel(): Unit = {

  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
    println("~~~run~~~~")
    DBs.setupAll()  // parse configuration file

    DB.readOnly{ implicit session => {
      SQL("select * from student").map(rs => {
        val student = Student(rs.int("id"),rs.string("name"),rs.string("password"),rs.int("age"))
        ctx.collect(student)
      }).list().apply()
    }

    }

  }
}

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

     env.addSource(new ScalikeJDBCMySQLSource).setParallelism(2).print()

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

结果是：
~~~run~~~~
~~~run~~~~
6> Student(1,kairis,wsx111,17)
1> Student(4,happy,11,48)
8> Student(3,double,44,12)
7> Student(2,dbh,11,90)
7> Student(3,double,44,12)
5> Student(1,kairis,wsx111,17)
6> Student(2,dbh,11,90)
8> Student(4,happy,11,48)

2并行度 也就是结果会拿取两份

第二个问题：
接着最开始流connect的问题 

1.log流 的并行度是4   （100w条）
那么 MySQL里的并行度是否有必要也是4呢？ （MySQl 1w条）


结果：
 是不是 没有这个必要呀 
 1个并行度能处理过来的 就可以了 
 但是1个并行度 也log流4个并行度 connect的时候还会出问题
那么该怎么办呢？之后说

输出对比：
Spark读写：外部数据源
    spark.read.format("").option("",").load
    spark.write.format("")....save

Flink读写
    addSource(new XXXSourceFunction)
    addSink(new XXXSinkFunction)

   都是  可插拔的  Spark 自定义外部数据源 
还是Spark好用呀

Flink的角色是Producer：

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

        val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
          val splits = x.split(",")
          Access(splits(0).toLong, splits(1), splits(2).toLong).toString
        })

        val producer = new FlinkKafkaProducer[String](
          "hadoop101:9092,hadoop101:9093,hadoop101:9094",         // broker list
          "double_happy_offset",               // target topic
          new SimpleStringSchema)   // serialization schema

        stream.addSink(producer)  // 2Kafka
        stream.print() // 2Local


    env.execute(this.getClass.getSimpleName)
  }
}
结果是：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic double_happy_offset 
Access(201912120010,ruozedata.com,2000)
Access(201912120010,ruozedata.com,4000)
Access(201912120010,dongqiudi.com,1000)
Access(201912120010,dongqiudi.com,6000)
Access(201912120010,zhibo8.com,5000)

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers",  "hadoop101:9092,hadoop101:9093,hadoop101:9094")
    properties.setProperty("group.id", "sxwang")
    
    val consumer = new FlinkKafkaConsumer[String]("double_happy_offset", new SimpleStringSchema(), properties)
    
    val stream = env.addSource(consumer)

    // TODO... Kafka2Kafka  double_happy_offset ==> double_happy_offset_test
    
    val producer = new FlinkKafkaProducer[String](
      "hadoop101:9092,hadoop101:9093,hadoop101:9094", // broker list
      "double_happy_offset_test", // target topic
      new SimpleStringSchema) // serialization schema

    stream.addSink(producer) // 2Kafka
    stream.print() // 2Local
    
    env.execute(this.getClass.getSimpleName)
  }

}

结果是：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic double_happy_offset_test 
Access(201912120010,ruozedata.com,2000)
Access(201912120010,dongqiudi.com,1000)
Access(201912120010,dongqiudi.com,6000)
e
c
Access(201912120010,ruozedata.com,4000)
b
e
Access(201912120010,zhibo8.com,5000)
c
f

/**
 * A {@link org.apache.flink.api.common.functions.RichFunction} version of {@link SinkFunction}.
 */
@Public
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {

	private static final long serialVersionUID = 1L;
}



/**
 * Interface for implementing user defined sink functionality.
 *
 * @param <IN> Input type parameter.
 */
@Public
public interface SinkFunction<IN> extends Function, Serializable {


所以
IN： 表示 Input type parameter. 

表示sink到的 MySQL表里的 字段类型

class DoubleHappyMySQLSink  extends RichSinkFunction[(String,Int)]{

  var connection:Connection = _
  var insertPstmt:PreparedStatement = _
  var updatePstmt:PreparedStatement = _

  // 打开connection等
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    /**
      * 这块与Spark不一样的 Spark是一个批次往下写 
      * 
      * 而Flink是 来一条就往下写  
      * 
      * 所以 :
      *   这块创建两个 PreparedStatement 也可以创建一个 需要创建表的时候指定 key 即可 
      *   
      *   创建 两个简单些  ： 有就更新 没有就插入 
      */
    connection = MySQLUtils.getConnection()
    insertPstmt = connection.prepareStatement("insert into domain(domain,traffic) values (?,?)")
    updatePstmt = connection.prepareStatement("update domain set traffic=? where domain=?")
  }

  /**
    * 写数据
    */
  override def invoke(value: (String, Int), context: SinkFunction.Context[_]): Unit = {
    // TODO   insert update

    /**
      * 这块并没有写错哦  要看 你的sql语句的 ？ 走的哦 
      * 
      * 先更新 有就更新 无就插入
      */
    updatePstmt.setInt(1, value._2)
    updatePstmt.setString(2, value._1)
    updatePstmt.execute()

    if(updatePstmt.getUpdateCount == 0) {
      insertPstmt.setString(1, value._1)
      insertPstmt.setInt(2, value._2)
      insertPstmt.execute()
    }
  }


  // 释放资源
  override def close(): Unit = {
    super.close()
    if(insertPstmt != null) insertPstmt.close()
    if(updatePstmt != null) updatePstmt.close()
    if(connection != null) connection.close()
  }

}


注意：
这代码不严谨的

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      (splits(1), splits(2).toInt)
    }).keyBy(0).sum(1)

    stream.addSink(new DoubleHappyMySQLSink)


    env.execute(this.getClass.getSimpleName)
  }

}

结果是： 我执行了两遍Flink程序  发现 结果幂等性 ok
mysql> select * from domain;
+---------------+---------+
| domain        | traffic |
+---------------+---------+
| dongqiudi.com |    7000 |
| ruozedata.com |    6000 |
| zhibo8.com    |    5000 |
+---------------+---------+
3 rows in set (0.00 sec)

mysql> select * from domain;
+---------------+---------+
| domain        | traffic |
+---------------+---------+
| dongqiudi.com |    7000 |
| ruozedata.com |    6000 |
| zhibo8.com    |    5000 |
+---------------+---------+
3 rows in set (0.00 sec)

mysql>

注意：
	你调试的时候 ：
		1.不需要接kafka   使用 自定义Source 就可以 eg：RichParallelSourceFunction
		2.Sink的时候 也不需要Sink 到 存储的地方去
			eg：kudu 、Hbase等
			这里使用 Iterator Data Sink
这个真的很重要：
	我封闭开发的时候 字段154个 Sink到kudu里面 
如果使用 提供的 connecer 154个字段 你什么时候能写完 
	一定是要写成 自动解析的 最好 

注意：
	刚好 我的业务场景 需要这个 Iterator Data Sink

官网：
import org.apache.flink.streaming.experimental.DataStreamUtils
import scala.collection.JavaConverters.asScalaIteratorConverter

val myResult: DataStream[(String, Int)] = ...
val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.javaStream).asScala

object SourceApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //测试使用

    val data: DataStream[Domain.Access] = env.addSource(new AccessSource03).setParallelism(1)

    import scala.collection.JavaConverters.asScalaIteratorConverter
    val output: Iterator[Domain.Access] = DataStreamUtils.collect(data.javaStream).asScala
    
    output.take(1).foreach(println(_))

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

结果：
Access(1577975577015,zhibo8.cc,78)

注意：
pom 选择的 版本;

<dependency>
  <groupId>org.apache.bahir</groupId>
  <artifactId>flink-connector-redis_2.11</artifactId>
  <version>1.1-SNAPSHOT</version>
</dependency>

这个版本可能下载不到 ：

所以下载：
<!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis -->
<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>

class DoubleHappyRedisSink  extends RedisMapper[(String,Int)]{
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "double_happy_traffic")
  }

  override def getValueFromData(data: (String, Int)): String = {
    data._2 + ""
  }

  override def getKeyFromData(data: (String, Int)): String = {
    data._1
  }
}

object SinkApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("C:\\IdeaProjects\\flink\\data\\access.log").map(x => {
      val splits = x.split(",")
      (splits(1), splits(2).toInt)
    }).keyBy(0).sum(1)


    val conf = new FlinkJedisPoolConfig.Builder().setHost("hadoop101").build()

    stream.addSink(new RedisSink[(String, Int)](conf, new DoubleHappyRedisSink))

    env.execute(this.getClass.getSimpleName)
  }

}
结果是：
hadoop101:6379> HGETALL double_happy_traffic
1) "ruozedata.com"
2) "6000"
3) "dongqiudi.com"
4) "7000"
5) "zhibo8.com"
6) "5000"
hadoop101:6379> 

与前面的MySQL 结果是一样的