使用Flume采集流式数据发送到Kafka，再由Flink消费Kafka数据，实现电话号码统计

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

//创建flink执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
//设置并行度
env.setParallelism(1)
//设置时间语义
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val properties = new Properties()
properties.setProperty("bootstrap.servers", "hdp111:9092,hdp222:9092,hdp333:9092")
properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
properties.setProperty("group.id", "test")
val dataStream = env.addSource(new FlinkKafkaConsumer[String]("mykafka", new SimpleStringSchema(), properties))
//val dataStream = env.readTextFile("C:\\Intel\\day0411\\input\\ww.txt")
.map(data=>{
val strings = data.split("\t")
LongEv(strings(0).toLong,strings(1).toLong)
}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LongEv](Time.seconds(1)) {
override def extractTimestamp(element: LongEv): Long = element.elTime*1000
})

// 6. 统计出5s内电话号码出现的总次数并打印到控制台
// dataStream.map(x=>(“电话号码总数”,1)).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print(“五秒内出现的总次数”)
// 7. 统计出5s内电话号码是偶数的电话并打印到控制台
// dataStream.filter(.phone % 2 == 0).map(x=>(x.phone, 1)).timeWindowAll(Time.seconds(5)).sum(1).print(“五秒内出现的偶数车牌”)
// 8. 统计出10s内同一电话号码出现次数超过2次及以上的电话号码并打印到控制台
// dataStream.map(x=>(x.phone,1)).keyBy(.1).timeWindow(Time.seconds(10)).sum(1).filter(._2>=2).print(“手机号出现次数超过2的”)
env.execute()
}