Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[v0.7.0] Redisgn Data Generator and External Kafka Source Support (#131)
* Improved Data Generator (#126) * Integrated index generation in DataGenerator; fix #119 * [Storm] Implemented new DataGenerator * [v0.6.1-dev] bump development version * [Flink] Implemented new data generator * [Storm] Implemented new data generator * [Flink] Implemented new data generator * [All] Fixed conversion from bytes to string * [All] Cleaned code * Update README.md * Fixed variable naming issue * Fixed DataGenerator null pointer exception * Sobstituded with correct conf file * [v0.7.0] External Source Support: Kafka (#130) * Added external Kafka Source with basic configurations (curr. only workflow schema) * Fix: added control for shape and external source * Added external Kafka Source also for Pipeline Schema * Added debugging output in source * cleaned code * [v0.7.0] External Source Support: Kafka
- Loading branch information
Showing
15 changed files
with
476 additions
and
55 deletions.
There are no files selected for viewing
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
84 changes: 84 additions & 0 deletions
84
...marks/flink-bench/src/main/java/fr/unice/namb/flink/utils/KafkaDeserializationSchema.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,84 @@ | ||
| package fr.unice.namb.flink.utils; | ||
|
|
||
| import org.apache.flink.api.common.serialization.DeserializationSchema; | ||
| import org.apache.flink.api.common.serialization.SerializationSchema; | ||
| import org.apache.flink.api.common.typeinfo.TypeHint; | ||
| import org.apache.flink.api.common.typeinfo.TypeInformation; | ||
| import org.apache.flink.api.java.tuple.Tuple4; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.ObjectOutputStream; | ||
| import java.nio.charset.Charset; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.UUID; | ||
|
|
||
| import static org.apache.flink.util.Preconditions.checkNotNull; | ||
|
|
||
| public class KafkaDeserializationSchema implements DeserializationSchema<Tuple4<String, String, Long, Long>>, SerializationSchema<String> { | ||
| private static final long serialVersionUID = 1L; | ||
|
|
||
| private transient Charset charset; | ||
| private long counter = 0; | ||
| private int rate; | ||
| private String me; | ||
|
|
||
|
|
||
| public KafkaDeserializationSchema(double frequency, String me) { | ||
| this.charset = checkNotNull(StandardCharsets.UTF_8); | ||
| this.me = me; | ||
| if (frequency > 0) this.rate = (int) (1/frequency); | ||
| } | ||
|
|
||
| public Charset getCharset() { | ||
| return charset; | ||
| } | ||
|
|
||
| // ------------------------------------------------------------------------ | ||
| // Kafka Serialization | ||
| // ------------------------------------------------------------------------ | ||
|
|
||
| @Override | ||
| public Tuple4<String, String, Long, Long> deserialize(byte[] message) { | ||
|
|
||
| String nextValue = new String(message, charset); | ||
| String tuple_id = UUID.randomUUID().toString(); | ||
| ++counter; | ||
| long ts = System.currentTimeMillis(); | ||
|
|
||
| if (this.rate > 0 && this.counter % this.rate == 0){ | ||
| System.out.println("[DEBUG] [" + this.me + "] : " + tuple_id + "," + this.counter + "," + ts + "," + nextValue); | ||
| } | ||
|
|
||
| return new Tuple4<>(nextValue, tuple_id, this.counter, ts); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isEndOfStream(Tuple4<String, String, Long, Long> nextElement) { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] serialize(String element) { | ||
| return element.getBytes(charset); | ||
| } | ||
|
|
||
| @Override | ||
| public TypeInformation<Tuple4<String, String, Long, Long>> getProducedType() { | ||
| return TypeInformation.of(new TypeHint<Tuple4<String, String, Long, Long>>(){}); | ||
| } | ||
|
|
||
| // ------------------------------------------------------------------------ | ||
| // Java Serialization | ||
| // ------------------------------------------------------------------------ | ||
|
|
||
| private void writeObject (ObjectOutputStream out) throws IOException { | ||
| out.defaultWriteObject(); | ||
| out.writeUTF(charset.name()); | ||
| } | ||
|
|
||
| private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException { | ||
| in.defaultReadObject(); | ||
| String charsetName = in.readUTF(); | ||
| this.charset = Charset.forName(charsetName); | ||
| } | ||
| } |
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
Oops, something went wrong.