本篇內容主要講解“flink怎么使用Event_time處理實時數據”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“flink怎么使用Event_time處理實時數據”吧!

讓客戶滿意是我們工作的目標，不斷超越客戶的期望值來自于我們對這個行業的熱愛。我們立志把好的技術通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領域值得信任、有價值的長期合作伙伴，公司提供的服務項目有：域名與空間、網絡空間、營銷軟件、網站建設、賈汪網站維護、網站推廣。

//flink中關于時間的三個概念
//event time:數據產生的時間
//processing time:處理數據的時間 即操作數據的之間 比如一個flink在scala中的map函數處理數據時
//ingest time:攝取數據時間，在一個streaming程序中 一個時間段收集數據的時間
//而evet time在處理實時數據時是比較有用的，例如在由于網絡的繁忙的原因，某些數據未能按時到達，假設遲到了30min,
//而我們定義的最大延遲不能超過十分鐘，那么一些數據包含了超時的數據那么這些數據是不會在這次操作中處理的而是會
//丟棄掉

//kafka生產者代碼
package kafka.partition.test;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class PartitionProducer {
	public static void main(String[] args) {
		
		Map<String,Object> props = new HashMap<>();
		props.put("acks", "1");
		props.put("partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("bootstrap.servers", "bigdata01:9092");
		String topic = "event_time";
		
		KafkaProducer<String, String> producer = new KafkaProducer<>(props);
		for(int i = 0 ; i <= 20;i++) {	
		//flink的watermarkassginer里面定義的超時時間是5000毫秒
			long mills = System.currentTimeMillis();
			if(i%3==0) {
			//數據的event time放在字符串的開頭 以空格分割
			//kafka event_time topic的0分區超時4000毫秒
				String line = (mills-4000)+" "+"partition-0--this is a big +" +i;
				ProducerRecord< String,String> record = new ProducerRecord<String, String>(topic, new Integer(0), null, i+"", line);
				producer.send(record);
			}else if(i%3==1) {
			//kafka event_time topic的1分區超時5000毫秒
				String line = (mills-5000)+" "+"partition-1--this is a big +" +i;
				ProducerRecord< String,String> record = new ProducerRecord<String, String>(topic, new Integer(1), null, i+"", line);
				producer.send(record);
			}else if(i%3==2) {
			//kafka event_time topic的2分區超時8000毫秒
				String line = (mills-8000)+" "+"partition-2--this is a big +" +i;
				ProducerRecord< String,String> record = new ProducerRecord<String, String>(topic, new Integer(2), null, i+"", line);
				producer.send(record);
			}
		}
		
		producer.close();
	}
}

//自定義的TimestampsAndWatermarks
package flink.streaming
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark
class CustomWaterMarks extends AssignerWithPeriodicWatermarks[String]{
    //超時時間
  val maxOutOrderness = 5000l
  //flink過一段時間便會調一次該函數獲取水印
  def getCurrentWatermark():Watermark ={
    val  sysMilssecons =  System.currentTimeMillis()
     new Watermark(sysMilssecons-maxOutOrderness) 
    
  }
  //每條記錄多會調用 來獲得even time 在生產的數據中 even_time放在字符串的第一個字段 用空格分割
  def extractTimestamp(element: String,previousElementTimestamp: Long): Long = {
   ((element.split(" ")).head).toLong
  }
}

package flink.streaming
import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
object StreamWithEventTimeAndWaterMarks {
  
  def main(args: Array[String]): Unit = {
    val kafkaProps = new Properties()
    //kafka的一些屬性
    kafkaProps.setProperty("bootstrap.servers", "bigdata01:9092")
    //所在的消費組
    kafkaProps.setProperty("group.id", "group2")
    //獲取當前的執行環境
    val evn = StreamExecutionEnvironment.getExecutionEnvironment
    //配制處理數據的時候使用event time
    evn.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //kafka的consumer，test1是要消費的topic
    val kafkaSource = new FlinkKafkaConsumer[String]("event_time",new SimpleStringSchema,kafkaProps)
    //添加自定義的 TimestampsAndWatermarks
    kafkaSource.assignTimestampsAndWatermarks(new CustomWaterMarks)
    //設置從最新的offset開始消費
    //kafkaSource.setStartFromGroupOffsets()
    kafkaSource.setStartFromLatest()
    //自動提交offset
    kafkaSource.setCommitOffsetsOnCheckpoints(true)
    
    //flink的checkpoint的時間間隔
    //evn.enableCheckpointing(2000)
    //添加consumer
    val stream = evn.addSource(kafkaSource)
    evn.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE)
    //stream.setParallelism(3)
    val text = stream.flatMap{ _.toLowerCase().split(" ").drop(1).filter { _.nonEmpty} }
          .map{(_,1)}
          .keyBy(0)
          .timeWindow(Time.seconds(5))
          .sum(1)
          .map(x=>{(x._1,(new Integer(x._2)))})
     text.print()
     //啟動執行    
     
     //text.addSink(new Ssinks())
     
    evn.execute("kafkawd")  
    
  }
  
}

打印結果 partition-2中的數據因為超時沒有出現
1> (big,14)
4> (is,14)
1> (+0,1)
2> (+1,1)
3> (partition-1--this,7)
4> (+15,1)
3> (+12,1)
1> (partition-0--this,7)
3> (+6,1)
1> (+16,1)
4> (+10,1)
2> (+18,1)
4> (+7,1)
3> (+3,1)
2> (+9,1)
3> (+19,1)
2> (+13,1)
3> (a,14)
2> (+4,1)

到此，相信大家對“flink怎么使用Event_time處理實時數據”有了更深的了解，不妨來實際操作一番吧！這里是創新互聯網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！

本文標題：flink怎么使用Event_time處理實時數據
文章轉載：http://m.newbst.com/article2/pjccoc.html

成都網站建設公司_創新互聯，為您提供用戶體驗、品牌網站設計、商城網站、微信公眾號、、動態網站

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯