Flink原理简介和使用(3)

a 、
Flink采用一种成为任务链的优化技术
可以在指定条件下减少本地通信开销
为了满足任务链的要求 
必须将两个或多个算子设为相同的并行度
并通过本地转发的方式连接(local forward)
b、
相同并行度one to one操作 Flink这样相连的算子链接在一起形成一个task 原来的算子成为里面的subtask
c、
并行度相同、并且是One-to-One操作 两个条件缺一不可
d、
将算子链接成task是非常有效的优化 
它能减少线程之间的切换和基于缓存区的数据交换
在减少时延的同时提升吞吐量 
链接的行为可以在编程API中进行指定

创建一个执行环境 表示当前执行程序的上下文
如果程序是独立调用的 则此方法返回本地执行环境
如果从命令行调用程序以提交到集群 则此方法返回集群的执行环境

val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
或
val env = StreamExecutionEnvironment.getExecutionEnvironment
a 返回本地执行环境 需要在调用时指定默认的并行度
val env = StreamExecutionEnvironment.createLocalEnvironment(1)
b 返回集群环境 将jar包提交到远程服务器 需要在调用时指定 JobManager的IP和端口号 并指定要在集群中运行的jar包
val env = ExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname",
6123,"YOURPATH//wordcount.jar")

如果没有设置并行度 会以flink-conf.yaml中配置为准 默认是1

//  定义样例类，传感器 id ，时间戳，温度
case class SensorReading(id: String, timestamp: Long, temperature: Double)
object Sensor {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val stream1 = env
          .fromCollection(List(
            SensorReading("sensor_1", 1547718199, 35.80018327300259),
            SensorReading("sensor_6", 1547718201, 15.402984393403084),
            SensorReading("sensor_7", 1547718202, 6.720945201171228),
            SensorReading("sensor_10", 1547718205, 38.101067604893444)
          ))
        stream1.print("stream1:").setParallelism(1)
        env.execute()
     }
}

https://archive.apache.org/dist/kafka/0.11.0.1/kafka_2.12-0.11.0.1.tgz

/opt/kafka/版本号/config/server.properties 这个配置文件
配置zk集群的配置项名称
kafka_2.10-0.8.2.1: zookeeper.contact
kafka_2.12-0.11.0.1: zookeeper.connect

./kafka-console-producer.sh --broker-list 192.168.84.128:9092 --topic test

val stream2 = env.readTextFile("YOUR_FILE_PATH")

需要引入kafka连接器的依赖 pom：
<dependency>
 <groupId>org.apache.flink</ groupId>
 <artifactId>flink-connector-kafka-0.11_2.11</artifactI   d>
 <version>1.7.2</version>
</dependency>
代码：
val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("group.id", "consumer-group")
properties.setProperty("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("auto.offset.reset", "latest")
val stream3 = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new
SimpleStringSchema(), properties))

cd /opt/kafka/kafka_2.10-0.8.2.1/bin
生产消息
./kafka-console-producer.sh --broker-list 192.168.84.128:9092 --topic test
消费消息
./kafka-console-consumer.sh --zookeeper 192.168.84.128:2181 --topic test --from-beginning

https://gitee.com/pingfanrenbiji/Flink-UserBehaviorAnalysis/blob/master/FlinkTutorial/src/main/scala/co