forked from phil3k3/flink-esper
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
moved Esper Engine to value state, started with type serialization
- Loading branch information
phil3k
committed
Oct 31, 2017
1 parent
73c71ae
commit ef827b5
Showing
5 changed files
with
237 additions
and
59 deletions.
There are no files selected for viewing
88 changes: 88 additions & 0 deletions
88
src/main/java/at/datasciencelabs/EsperEngineSerializer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package at.datasciencelabs; | ||
|
|
||
| import com.espertech.esper.client.EPServiceProvider; | ||
| import org.apache.flink.api.common.typeutils.CompatibilityResult; | ||
| import org.apache.flink.api.common.typeutils.TypeSerializer; | ||
| import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot; | ||
| import org.apache.flink.core.memory.DataInputView; | ||
| import org.apache.flink.core.memory.DataOutputView; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| public class EsperEngineSerializer extends TypeSerializer<EPServiceProvider> { | ||
|
|
||
| @Override | ||
| public boolean isImmutableType() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public TypeSerializer<EPServiceProvider> duplicate() { | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public EPServiceProvider createInstance() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public EPServiceProvider copy(EPServiceProvider from) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public EPServiceProvider copy(EPServiceProvider from, EPServiceProvider reuse) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public int getLength() { | ||
| return -1; | ||
| } | ||
|
|
||
| @Override | ||
| public void serialize(EPServiceProvider record, DataOutputView target) throws IOException { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public EPServiceProvider deserialize(DataInputView source) throws IOException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public EPServiceProvider deserialize(EPServiceProvider reuse, DataInputView source) throws IOException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public void copy(DataInputView source, DataOutputView target) throws IOException { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object obj) { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean canEqual(Object obj) { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public TypeSerializerConfigSnapshot snapshotConfiguration() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public CompatibilityResult<EPServiceProvider> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) { | ||
| return null; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,67 +1,92 @@ | ||
| package at.datasciencelabs; | ||
|
|
||
| import org.apache.flink.api.common.typeinfo.TypeInformation; | ||
| import com.espertech.esper.client.EventBean; | ||
| import org.apache.flink.streaming.api.datastream.DataStream; | ||
| import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; | ||
| import org.apache.flink.streaming.api.functions.sink.SinkFunction; | ||
| import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase; | ||
| import org.junit.Before; | ||
| import org.junit.Test; | ||
|
|
||
| public class EsperStreamTest { | ||
| import java.io.Serializable; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| @Test | ||
| public void testEsperStream() throws Exception { | ||
|
|
||
| StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(); | ||
| import static org.hamcrest.CoreMatchers.is; | ||
| import static org.hamcrest.MatcherAssert.assertThat; | ||
| import static org.hamcrest.core.IsNull.notNullValue; | ||
|
|
||
| DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25), new TestEvent("maria", 30)); | ||
|
|
||
| EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent"); | ||
|
|
||
| DataStream<TestEvent> resultStream = esperStream.select((EsperSelectFunction<TestEvent>) collector -> { | ||
| String name = (String) collector.get("name"); | ||
| int age = (int) collector.get("age"); | ||
| return new TestEvent(name, age); | ||
| }, TypeInformation.of(TestEvent.class)); | ||
| public class EsperStreamTest extends StreamingMultipleProgramsTestBase implements Serializable { | ||
|
|
||
| resultStream.printToErr(); | ||
| private static List<TestEvent> result; | ||
|
|
||
| executionEnvironment.execute("test"); | ||
|
|
||
| @Before | ||
| public void before() { | ||
| result = new ArrayList<>(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testMoreComplexEsperStream() throws Exception { | ||
| @SuppressWarnings("Convert2Lambda") | ||
| public void shouldSelectFromStreamUsingAnonymousClassSelect() throws Exception { | ||
| StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(); | ||
| executionEnvironment.setParallelism(1); | ||
|
|
||
| DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25), new TestEvent("maria", 30)); | ||
|
|
||
| EsperStream<TestEvent> eventEsperStream = new EsperStream<>(dataStream, "select avg(age) as average_age from TestEvent"); | ||
|
|
||
| DataStream<Double> resultStream = eventEsperStream.select((EsperSelectFunction<Double>) collector -> { | ||
| Double age = (Double) collector.get("average_age"); | ||
| return age; | ||
| }, TypeInformation.of(Double.class)); | ||
|
|
||
| resultStream.printToErr(); | ||
| EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent"); | ||
|
|
||
| executionEnvironment.execute("test"); | ||
| DataStream<TestEvent> resultStream = esperStream.select(new EsperSelectFunction<TestEvent>() { | ||
| @Override | ||
| public TestEvent select(EventBean eventBean) throws Exception { | ||
| String name = (String) eventBean.get("name"); | ||
| int age = (int) eventBean.get("age"); | ||
| return new TestEvent(name, age); | ||
| } | ||
| }); | ||
|
|
||
| resultStream.addSink(new SinkFunction<TestEvent>() { | ||
| @Override | ||
| public void invoke(TestEvent testEvent) throws Exception { | ||
| System.err.println(testEvent); | ||
| result.add(testEvent); | ||
| } | ||
| }); | ||
|
|
||
| executionEnvironment.execute("test-2"); | ||
|
|
||
| assertThat(result, is(notNullValue())); | ||
| assertThat(result.size(), is(3)); | ||
| } | ||
|
|
||
| @Test | ||
| public void testStateChangeRule() throws Exception { | ||
| @SuppressWarnings("Convert2Lambda") | ||
| public void shouldSelectFromStreamUsingLambdaSelect() throws Exception { | ||
|
|
||
| StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(); | ||
| executionEnvironment.setParallelism(1); | ||
|
|
||
| DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter1", 10), new TestEvent("alex1", 25), new TestEvent("maria1", 30)); | ||
|
|
||
| DataStream<TestEvent> dataStream = executionEnvironment.fromElements(new TestEvent("peter", 10), new TestEvent("alex", 25)); | ||
| EsperStream<TestEvent> esperStream = new EsperStream<>(dataStream, "select name, age from TestEvent"); | ||
|
|
||
| EsperStream<TestEvent> eventEsperStream = new EsperStream<>(dataStream, "select avg(age) as average_age from TestEvent"); | ||
| DataStream<TestEvent> resultStream = esperStream.select((EsperSelectFunction<TestEvent>) collector -> { | ||
| String name = (String) collector.get("name"); | ||
| int age = (int) collector.get("age"); | ||
| return new TestEvent(name, age); | ||
| }); | ||
|
|
||
| DataStream<Double> resultStream = eventEsperStream.select((EsperSelectFunction<Double>) collector -> { | ||
| Double age = (Double) collector.get("average_age"); | ||
| return age; | ||
| }, TypeInformation.of(Double.class)); | ||
| resultStream.addSink(new SinkFunction<TestEvent>() { | ||
| @Override | ||
| public void invoke(TestEvent testEvent) throws Exception { | ||
| result.add(testEvent); | ||
| } | ||
| }); | ||
|
|
||
| resultStream.printToErr(); | ||
| executionEnvironment.execute("test-1"); | ||
|
|
||
| executionEnvironment.execute("stateChange1"); | ||
| assertThat(result, is(notNullValue())); | ||
| assertThat(result.size(), is(3)); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.