我在用Flink读动觉流。它根据时间窗口和键聚合特定事件。代码在reduce之后不执行任何操作。输出csv中未Map任何数据。我已经等了很多分钟(即使时间窗口只有两分钟)。
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();
}
};
1条答案
按热度按时间9nvpjoqh1#
这是个愚蠢的错误。
调用返回一个带有指定水印的新流。这个返回值应该在以后的操作中使用。