使用Flink和kinesis流的流窗口處理處理不起作用

我正在使用Flink讀取kinesis流。它基於時間窗口和密鑰聚合某個事件。代碼在reduce之後不做任何事情。沒有數據被映射到輸出csv中。我等了很多分鐘（即使時間窗口只有兩分鐘）。使用Flink和kinesis流的流窗口處理處理不起作用

public static void main(String[] args) throws Exception { 

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
    env.enableCheckpointing(CommonTimeConstants.TWO_MINUTES.toMilliseconds()); 
    env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); 
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(1, TimeUnit.MINUTES))); 

    Properties consumerConfig = new Properties(); 
    consumerConfig.put(ConsumerConfigConstants.AWS_REGION, PropertyFileUtils.get("aws.region", "")); 
    consumerConfig.put(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, PropertyFileUtils.get("aws.accessKeyId", "")); 
    consumerConfig.put(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, PropertyFileUtils.get("aws.secretAccessKey", "")); 
    consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON"); 

    DataStream<APIActionLog> apiLogRecords = env.addSource(new FlinkKinesisConsumer<>(
      ProjectProperties.SOURCE_ENV_PREFIX, // stream name 
      new StreamedApiLogRecordDeserializationSchema(), 
      consumerConfig)); 

    apiLogRecords.assignTimestampsAndWatermarks(API_LOG_RECORD_BOUNDED_OUT_OF_ORDERNESS_TIMESTAMP_EXTRACTOR); 

    DataStream<Tuple7<String, String, String, String, Timestamp, String, Integer>> skuPlatformTsCount = 
      apiLogRecords.flatMap(collecting events...) 
        .keyBy(Key based on some parameters of the event...) 
        .timeWindow(TWO_MINUTES) 
        .reduce(adding up event parameter..., window function...) 
        .map(Map to get a different tuple format...); 

    skuPlatformTsCount.writeAsCsv("/Users/uday/Desktop/out.csv", FileSystem.WriteMode.OVERWRITE); 

    env.execute("Processing ATC Log Stream"); 
} 

private static final BoundedOutOfOrdernessTimestampExtractor<APIActionLog> API_LOG_RECORD_BOUNDED_OUT_OF_ORDERNESS_TIMESTAMP_EXTRACTOR = 
     new BoundedOutOfOrdernessTimestampExtractor<APIActionLog>(TEN_SECONDS) { 
      private static final long serialVersionUID = 1L; 

      @Override 
      public long extractTimestamp(APIActionLog apiActionLog) { 
       return apiActionLog.getTs().getTime(); 
      } 
     };

來源

2017-08-28 uday

這是一個愚蠢的錯誤。

apiLogRecords.assignTimestampsAndWatermarks(API_LOG_RECORD_BOUNDED_OUT_OF_ORDERNESS_TIMESTAMP_EXTRACTOR);

調用返回一個帶有分配水印的新流。這個返回的值應該在以後的操作中使用。

來源

2017-08-30 09:57:52 uday

使用Flink和kinesis流的流窗口處理處理不起作用

回答

相關問題