Skip to content

Commit 2aebabd

Browse files
msprucmspruc
committed
Source, map & collection sink
1 parent f7dff1e commit 2aebabd

12 files changed

Lines changed: 742 additions & 122 deletions

File tree

‎wayang-platforms/wayang-flink/pom.xml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,16 @@
116116
<artifactId>flink-hadoop-compatibility_2.12</artifactId>
117117
<version>${flink.version}</version>
118118
</dependency>
119+
<dependency>
120+
<groupId>org.apache.flink</groupId>
121+
<artifactId>flink-connector-files</artifactId>
122+
<version>${flink.version}</version>
123+
</dependency>
124+
<dependency>
125+
<groupId>org.apache.flink</groupId>
126+
<artifactId>flink-streaming-java</artifactId>
127+
<version>${flink.version}</version>
128+
</dependency>
119129
<dependency>
120130
<groupId>org.apache.commons</groupId>
121131
<artifactId>commons-math3</artifactId>

‎wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/channels/DataStreamChannel.java‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.wayang.flink.channels;
20+
21+
import org.apache.flink.streaming.api.datastream.DataStream;
22+
import org.apache.wayang.core.optimizer.OptimizationContext;
23+
import org.apache.wayang.core.plan.executionplan.Channel;
24+
import org.apache.wayang.core.plan.wayangplan.OutputSlot;
25+
import org.apache.wayang.core.platform.AbstractChannelInstance;
26+
import org.apache.wayang.core.platform.ChannelDescriptor;
27+
import org.apache.wayang.core.platform.ChannelInstance;
28+
import org.apache.wayang.core.platform.Executor;
29+
import org.apache.wayang.flink.execution.FlinkExecutor;
30+
31+
import java.util.OptionalLong;
32+
33+
public class DataStreamChannel extends Channel {
34+
35+
/**
36+
* {@link ChannelInstance} implementation for {@link DataStream}s.
37+
*/
38+
public class Instance extends AbstractChannelInstance {
39+
40+
private DataStream<?> dataStream;
41+
42+
// TODO: this.size is currently always 0
43+
private long size;
44+
45+
public Instance(final FlinkExecutor executor,
46+
final OptimizationContext.OperatorContext producerOperatorContext,
47+
final int producerOutputIndex) {
48+
super(executor, producerOperatorContext, producerOutputIndex);
49+
}
50+
51+
public void accept(final DataStream<?> dataStream) {
52+
this.dataStream = dataStream;
53+
}
54+
55+
@SuppressWarnings("unchecked")
56+
public <T> DataStream<T> provideDataStream() {
57+
return (DataStream<T>) this.dataStream;
58+
}
59+
60+
@Override
61+
public OptionalLong getMeasuredCardinality() {
62+
return this.size == 0 ? super.getMeasuredCardinality() : OptionalLong.of(this.size);
63+
}
64+
65+
@Override
66+
public DataStreamChannel getChannel() {
67+
return DataStreamChannel.this;
68+
}
69+
70+
@Override
71+
protected void doDispose() {
72+
this.dataStream = null;
73+
}
74+
}
75+
76+
public static final ChannelDescriptor DESCRIPTOR = new ChannelDescriptor(
77+
DataStreamChannel.class, true, false);
78+
79+
public static final ChannelDescriptor DESCRIPTOR_MANY = new ChannelDescriptor(
80+
DataStreamChannel.class, true, false);
81+
82+
public DataStreamChannel(final ChannelDescriptor descriptor, final OutputSlot<?> outputSlot) {
83+
super(descriptor, outputSlot);
84+
assert descriptor == DESCRIPTOR || descriptor == DESCRIPTOR_MANY;
85+
this.markForInstrumentation();
86+
}
87+
88+
private DataStreamChannel(final DataStreamChannel parent) {
89+
super(parent);
90+
}
91+
92+
@Override
93+
public Channel copy() {
94+
return new DataStreamChannel(this);
95+
}
96+
97+
@Override
98+
public Instance createInstance(final Executor executor,
99+
final OptimizationContext.OperatorContext producerOperatorContext,
100+
final int producerOutputIndex) {
101+
return new Instance((FlinkExecutor) executor, producerOperatorContext, producerOutputIndex);
102+
}
103+
}

‎wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/execution/FlinkExecutor.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.wayang.flink.execution;
2020

2121
import org.apache.flink.api.java.ExecutionEnvironment;
22+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
2223
import org.apache.wayang.core.api.Job;
2324
import org.apache.wayang.core.api.exception.WayangException;
2425
import org.apache.wayang.core.optimizer.OptimizationContext;
@@ -35,7 +36,6 @@
3536
import org.apache.wayang.flink.compiler.FunctionCompiler;
3637
import org.apache.wayang.flink.operators.FlinkExecutionOperator;
3738
import org.apache.wayang.flink.platform.FlinkPlatform;
38-
import com.esotericsoftware.kryo.serializers.DefaultSerializers;
3939

4040
import java.util.Arrays;
4141
import java.util.Collection;
@@ -56,6 +56,12 @@ public class FlinkExecutor extends PushExecutorTemplate {
5656
*/
5757
public ExecutionEnvironment fee;
5858

59+
60+
/**
61+
* {@link StreamExecutionEnvironment} for bounded and continuous streams.
62+
*/
63+
public StreamExecutionEnvironment sEnv;
64+
5965
/**
6066
* Compiler to create flink UDFs.
6167
*/

‎wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkBoundedTextFileSource.java‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.wayang.flink.operators;
20+
21+
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
22+
import org.apache.flink.connector.file.src.FileSource;
23+
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
24+
import org.apache.flink.core.fs.Path;
25+
import org.apache.flink.streaming.api.datastream.DataStream;
26+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
27+
28+
import org.apache.wayang.basic.operators.TextFileSource;
29+
import org.apache.wayang.core.optimizer.OptimizationContext.OperatorContext;
30+
import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators;
31+
import org.apache.wayang.core.platform.ChannelDescriptor;
32+
import org.apache.wayang.core.platform.ChannelInstance;
33+
import org.apache.wayang.core.platform.lineage.ExecutionLineageNode;
34+
import org.apache.wayang.core.util.Tuple;
35+
import org.apache.wayang.flink.channels.DataStreamChannel;
36+
import org.apache.wayang.flink.execution.FlinkExecutor;
37+
38+
import java.util.Arrays;
39+
import java.util.Collection;
40+
import java.util.List;
41+
42+
/**
43+
* Opens a Flink bounded {@link DataStream} on a {@link TextFileSource}.
44+
*/
45+
public class FlinkBoundedTextFileSource extends TextFileSource implements FlinkExecutionOperator {
46+
47+
public FlinkBoundedTextFileSource(final TextFileSource that) {
48+
super(that);
49+
}
50+
51+
public FlinkBoundedTextFileSource(final String inputUrl) {
52+
super(inputUrl);
53+
}
54+
55+
@Override
56+
public List<ChannelDescriptor> getSupportedInputChannels(final int index) {
57+
throw new UnsupportedOperationException(String.format("%s does not have input channels.", this));
58+
}
59+
60+
@Override
61+
public List<ChannelDescriptor> getSupportedOutputChannels(final int index) {
62+
return Arrays.asList(DataStreamChannel.DESCRIPTOR, DataStreamChannel.DESCRIPTOR_MANY);
63+
}
64+
65+
@Override
66+
public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> evaluate(final ChannelInstance[] inputs,
67+
final ChannelInstance[] outputs, final FlinkExecutor flinkExecutor, final OperatorContext operatorContext)
68+
throws Exception {
69+
assert inputs.length == this.getNumInputs();
70+
assert outputs.length == this.getNumOutputs();
71+
72+
final DataStreamChannel.Instance output = (DataStreamChannel.Instance) outputs[0];
73+
74+
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
75+
76+
final FileSource<String> fs = FileSource
77+
.forRecordStreamFormat(new TextLineInputFormat(), new Path(this.getInputUrl()))
78+
.build();
79+
80+
final DataStream<String> dataStream = env.fromSource(fs, WatermarkStrategy.noWatermarks(),
81+
"FlinkDataStreamFileSource[" + this.getInputUrl() + "]");
82+
83+
output.accept(dataStream);
84+
85+
final ExecutionLineageNode prepareLineageNode = new ExecutionLineageNode(operatorContext);
86+
prepareLineageNode.add(LoadProfileEstimators.createFromSpecification(
87+
"wayang.flink.textfilesource.load.prepare", flinkExecutor.getConfiguration()));
88+
89+
final ExecutionLineageNode mainLineageNode = new ExecutionLineageNode(operatorContext);
90+
mainLineageNode.add(LoadProfileEstimators.createFromSpecification(
91+
"wayang.flink.textfilesource.load.main", flinkExecutor.getConfiguration()));
92+
93+
output.getLineage().addPredecessor(mainLineageNode);
94+
95+
return prepareLineageNode.collectAndMark();
96+
}
97+
98+
@Override
99+
public boolean containsAction() {
100+
return false;
101+
}
102+
}

‎wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkContinuousTextFileSource.java‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.wayang.flink.operators;
20+
21+
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
22+
import org.apache.flink.connector.file.src.FileSource;
23+
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
24+
import org.apache.flink.core.fs.Path;
25+
import org.apache.flink.streaming.api.datastream.DataStream;
26+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
27+
28+
import org.apache.wayang.basic.operators.TextFileSource;
29+
import org.apache.wayang.core.optimizer.OptimizationContext.OperatorContext;
30+
import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators;
31+
import org.apache.wayang.core.platform.ChannelDescriptor;
32+
import org.apache.wayang.core.platform.ChannelInstance;
33+
import org.apache.wayang.core.platform.lineage.ExecutionLineageNode;
34+
import org.apache.wayang.core.util.Tuple;
35+
import org.apache.wayang.flink.channels.DataStreamChannel;
36+
import org.apache.wayang.flink.execution.FlinkExecutor;
37+
38+
import java.time.Duration;
39+
import java.util.Collection;
40+
import java.util.List;
41+
42+
/**
43+
* Opens a Flink continuous {@link DataStream} that monitors a file directory.
44+
*/
45+
public class FlinkContinuousTextFileSource extends TextFileSource implements FlinkExecutionOperator {
46+
47+
public FlinkContinuousTextFileSource(final TextFileSource that) {
48+
super(that);
49+
}
50+
51+
public FlinkContinuousTextFileSource(final String inputUrl) {
52+
super(inputUrl);
53+
}
54+
55+
@Override
56+
public List<ChannelDescriptor> getSupportedInputChannels(final int index) {
57+
throw new UnsupportedOperationException("Unimplemented method 'getSupportedInputChannels'");
58+
}
59+
60+
@Override
61+
public List<ChannelDescriptor> getSupportedOutputChannels(final int index) {
62+
throw new UnsupportedOperationException("Unimplemented method 'getSupportedOutputChannels'");
63+
}
64+
65+
@Override
66+
public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> evaluate(final ChannelInstance[] inputs,
67+
final ChannelInstance[] outputs, final FlinkExecutor flinkExecutor, final OperatorContext operatorContext)
68+
throws Exception {
69+
assert inputs.length == this.getNumInputs();
70+
assert outputs.length == this.getNumOutputs();
71+
72+
final DataStreamChannel.Instance output = (DataStreamChannel.Instance) outputs[0];
73+
74+
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
75+
76+
final FileSource<String> fs = FileSource
77+
.forRecordStreamFormat(new TextLineInputFormat(), new Path(this.getInputUrl()))
78+
//TODO: I set manually 1 here but should be in config.
79+
.monitorContinuously(Duration.ofSeconds(1))
80+
.build();
81+
82+
final DataStream<String> dataStream = env.fromSource(fs, WatermarkStrategy.noWatermarks(),
83+
"FlinkDataStreamFileSource[" + this.getInputUrl() + "]");
84+
85+
output.accept(dataStream);
86+
87+
final ExecutionLineageNode prepareLineageNode = new ExecutionLineageNode(operatorContext);
88+
prepareLineageNode.add(LoadProfileEstimators.createFromSpecification(
89+
"wayang.flink.textfilesource.load.prepare", flinkExecutor.getConfiguration()));
90+
91+
final ExecutionLineageNode mainLineageNode = new ExecutionLineageNode(operatorContext);
92+
mainLineageNode.add(LoadProfileEstimators.createFromSpecification(
93+
"wayang.flink.textfilesource.load.main", flinkExecutor.getConfiguration()));
94+
95+
output.getLineage().addPredecessor(mainLineageNode);
96+
97+
return prepareLineageNode.collectAndMark();
98+
}
99+
100+
@Override
101+
public boolean containsAction() {
102+
throw new UnsupportedOperationException("Unimplemented method 'containsAction'");
103+
}
104+
105+
}

0 commit comments

Comments
 (0)