org.apache.flink.api.common.ExecutionConfig.setAutoWatermarkInterval()方法的使用及代码示例

本文整理了Java中org.apache.flink.api.common.ExecutionConfig.setAutoWatermarkInterval()方法的一些代码示例，展示了ExecutionConfig.setAutoWatermarkInterval()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ExecutionConfig.setAutoWatermarkInterval()方法的具体详情如下：
包路径：org.apache.flink.api.common.ExecutionConfig
类名称：ExecutionConfig
方法名：setAutoWatermarkInterval

ExecutionConfig.setAutoWatermarkInterval介绍

[英]Sets the interval of the automatic watermark emission. Watermarks are used throughout the streaming system to keep track of the progress of time. They are used, for example, for time based windowing.
[中]设置自动水印发射的间隔。整个流媒体系统都使用水印来跟踪时间的进程。例如，它们用于基于时间的窗口设置。

代码示例

代码示例来源：origin: apache/flink

/**
* Sets the time characteristic for all streams create from this environment, e.g., processing
* time, event time, or ingestion time.
*
*

If you set the characteristic to IngestionTime of EventTime this will set a default
* watermark update interval of 200 ms. If this is not applicable for your application
* you should change it using {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
*
* @param characteristic The time characteristic.
*/
@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}

代码示例来源：origin: apache/flink

executionConfig.setAutoWatermarkInterval(watermarkInterval);

代码示例来源：origin: apache/flink

sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
sEnv.enableCheckpointing(4000);
sEnv.getConfig().setAutoWatermarkInterval(1000);

代码示例来源：origin: apache/flink

@Test
public void testNegativeTimestamps() throws Exception {
final AssignerWithPeriodicWatermarks assigner = new NeverWatermarkExtractor();
final TimestampsAndPeriodicWatermarksOperator operator =
new TimestampsAndPeriodicWatermarksOperator(assigner);
OneInputStreamOperatorTestHarness testHarness =
new OneInputStreamOperatorTestHarness(operator);
testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
testHarness.open();
long[] values = { Long.MIN_VALUE, -1L, 0L, 1L, 2L, 3L, Long.MAX_VALUE };
for (long value : values) {
testHarness.processElement(new StreamRecord<>(value));
}
ConcurrentLinkedQueue output = testHarness.getOutput();
for (long value: values) {
assertEquals(value, ((StreamRecord) output.poll()).getTimestamp());
}
}

代码示例来源：origin: apache/flink

public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
String outputPath = params.getRequired("outputPath");
int recordsPerSecOnd= params.getInt("recordsPerSecond", 10);
int duration = params.getInt("durationInSecond", 60);
int offset = params.getInt("offsetInSecond", 0);
StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
sEnv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
sEnv.enableCheckpointing(4000);
sEnv.getConfig().setAutoWatermarkInterval(1000);
// execute a simple pass through program.
PeriodicSourceGenerator generator = new PeriodicSourceGenerator(
recordsPerSecond, duration, offset);
DataStream rows = sEnv.addSource(generator);
DataStream result = rows
.keyBy(1)
.timeWindow(Time.seconds(5))
.sum(0);
result.writeAsText(outputPath + "/result.txt", FileSystem.WriteMode.OVERWRITE)
.setParallelism(1);
sEnv.execute();
}

代码示例来源：origin: apache/flink

env.getConfig().setAutoWatermarkInterval(10);
env.setParallelism(1);
env.getConfig().disableSysoutLogging();

代码示例来源：origin: apache/flink

env.getConfig().setAutoWatermarkInterval(1);
env.setParallelism(2);
env.getConfig().disableSysoutLogging();

代码示例来源：origin: apache/flink

env.getConfig().setAutoWatermarkInterval(1);
env.setParallelism(1);
env.getConfig().disableSysoutLogging();

代码示例来源：origin: apache/flink

env.getConfig().setAutoWatermarkInterval(10);
env.setParallelism(2);
env.getConfig().disableSysoutLogging();

代码示例来源：origin: apache/flink

env.getConfig().setAutoWatermarkInterval(10);
env.setParallelism(1);
env.getConfig().disableSysoutLogging();

代码示例来源：origin: apache/flink

new OneInputStreamOperatorTestHarness<>(operator);
testHarness.getExecutionConfig().setAutoWatermarkInterval(50);

代码示例来源：origin: apache/flink

env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));

代码示例来源：origin: apache/flink

env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));

代码示例来源：origin: apache/flink

env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));

代码示例来源：origin: apache/flink

env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.11

/**
* Sets the time characteristic for all streams create from this environment, e.g., processing
* time, event time, or ingestion time.
*
*

If you set the characteristic to IngestionTime of EventTime this will set a default
* watermark update interval of 200 ms. If this is not applicable for your application
* you should change it using {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
*
* @param characteristic The time characteristic.
*/
@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}

代码示例来源：origin: DTStack/flinkx

/**
* Sets the time characteristic for all streams create from this environment, e.g., processing
* time, event time, or ingestion time.
*
*

If you set the characteristic to IngestionTime of EventTime this will set a default
* watermark update interval of 200 ms. If this is not applicable for your application
* you should change it using {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
*
* @param characteristic The time characteristic.
*/
@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.10

/**
* Sets the time characteristic for all streams create from this environment, e.g., processing
* time, event time, or ingestion time.
*
*

If you set the characteristic to IngestionTime of EventTime this will set a default
* watermark update interval of 200 ms. If this is not applicable for your application
* you should change it using {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
*
* @param characteristic The time characteristic.
*/
@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java

/**
* Sets the time characteristic for all streams create from this environment, e.g., processing
* time, event time, or ingestion time.
*
*

If you set the characteristic to IngestionTime of EventTime this will set a default
* watermark update interval of 200 ms. If this is not applicable for your application
* you should change it using {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
*
* @param characteristic The time characteristic.
*/
@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}

代码示例来源：origin: seznam/euphoria

env.getConfig().setAutoWatermarkInterval(autoWatermarkInterval.toMillis());