@PublicEvolving public class HybridSource<T> extends Object implements Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
A simple example with FileSource and KafkaSource with fixed Kafka start position:
FileSource<String> fileSource =
FileSource.forRecordStreamFormat(new TextLineInputFormat(), Path.fromLocalFile(testDir)).build();
KafkaSource<String> kafkaSource =
KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092")
.setGroupId("MyGroup")
.setTopics(Arrays.asList("quickstart-events"))
.setDeserializer(
KafkaRecordDeserializer.valueOnly(StringDeserializer.class))
.setStartingOffsets(OffsetsInitializer.earliest())
.build();
HybridSource<String> hybridSource =
HybridSource.builder(fileSource)
.addSource(kafkaSource)
.build();
A more complex example with Kafka start position derived from previous source:
HybridSource<String> hybridSource =
HybridSource.<String, StaticFileSplitEnumerator>builder(fileSource)
.addSource(
switchContext -> {
StaticFileSplitEnumerator previousEnumerator =
switchContext.getPreviousEnumerator();
// how to get timestamp depends on specific enumerator
long timestamp = previousEnumerator.getEndTimestamp();
OffsetsInitializer offsets =
OffsetsInitializer.timestamp(timestamp);
KafkaSource<String> kafkaSource =
KafkaSource.<String>builder()
.setBootstrapServers("localhost:9092")
.setGroupId("MyGroup")
.setTopics(Arrays.asList("quickstart-events"))
.setDeserializer(
KafkaRecordDeserializer.valueOnly(StringDeserializer.class))
.setStartingOffsets(offsets)
.build();
return kafkaSource;
},
Boundedness.CONTINUOUS_UNBOUNDED)
.build();
Modifier and Type | Class and Description |
---|---|
static class |
HybridSource.HybridSourceBuilder<T,EnumT extends SplitEnumerator>
Builder for HybridSource.
|
static interface |
HybridSource.SourceFactory<T,SourceT extends Source<T,?,?>,FromEnumT extends SplitEnumerator>
Factory for underlying sources of
HybridSource . |
static interface |
HybridSource.SourceSwitchContext<EnumT>
Context provided to source factory.
|
Modifier | Constructor and Description |
---|---|
protected |
HybridSource(List<org.apache.flink.connector.base.source.hybrid.HybridSource.SourceListEntry> sources)
Protected for subclass, use
builder(Source) to construct source. |
protected HybridSource(List<org.apache.flink.connector.base.source.hybrid.HybridSource.SourceListEntry> sources)
builder(Source)
to construct source.public static <T,EnumT extends SplitEnumerator> HybridSource.HybridSourceBuilder<T,EnumT> builder(Source<T,?,?> firstSource)
HybridSource
.public Boundedness getBoundedness()
Source
getBoundedness
in interface Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
@Internal public SourceReader<T,HybridSourceSplit> createReader(SourceReaderContext readerContext) throws Exception
SourceReaderFactory
createReader
in interface SourceReaderFactory<T,HybridSourceSplit>
readerContext
- The context
for the source reader.Exception
- The implementor is free to forward all exceptions directly. Exceptions
thrown from this method cause task failure/recovery.@Internal public SplitEnumerator<HybridSourceSplit,HybridSourceEnumeratorState> createEnumerator(SplitEnumeratorContext<HybridSourceSplit> enumContext)
Source
createEnumerator
in interface Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
enumContext
- The context
for the split enumerator.@Internal public SplitEnumerator<HybridSourceSplit,HybridSourceEnumeratorState> restoreEnumerator(SplitEnumeratorContext<HybridSourceSplit> enumContext, HybridSourceEnumeratorState checkpoint) throws Exception
Source
restoreEnumerator
in interface Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
enumContext
- The context
for the restored split
enumerator.checkpoint
- The checkpoint to restore the SplitEnumerator from.Exception
- The implementor is free to forward all exceptions directly. Exceptions
thrown from this method cause JobManager failure/recovery.@Internal public SimpleVersionedSerializer<HybridSourceSplit> getSplitSerializer()
Source
getSplitSerializer
in interface Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
@Internal public SimpleVersionedSerializer<HybridSourceEnumeratorState> getEnumeratorCheckpointSerializer()
Source
SplitEnumerator
checkpoint. The serializer is used for
the result of the SplitEnumerator.snapshotState(long)
method.getEnumeratorCheckpointSerializer
in interface Source<T,HybridSourceSplit,HybridSourceEnumeratorState>
Copyright © 2014–2024 The Apache Software Foundation. All rights reserved.