Java源码示例：org.apache.flink.streaming.util.serialization.TypeInformationKeyValueSerializationSchema
示例1
public void runKeyValueTest() throws Exception {
final String topic = "keyvaluetest";
createTestTopic(topic, 1, 1);
final int elementCount = 5000;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<Long, PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<Long, PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<Long, PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
PojoValue pojo = new PojoValue();
pojo.when = new Date(rnd.nextLong());
pojo.lon = rnd.nextLong();
pojo.lat = i;
// make every second key null to ensure proper "null" serialization
Long key = (i % 2 == 0) ? null : i;
ctx.collect(new Tuple2<>(key, pojo));
}
}
@Override
public void cancel() {
}
});
KeyedSerializationSchema<Tuple2<Long, PojoValue>> schema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write KV to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
KafkaDeserializationSchema<Tuple2<Long, PojoValue>> readSchema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<Long, PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, readSchema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<Long, PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<Long, PojoValue> value, Collector<Object> out) throws Exception {
// the elements should be in order.
Assert.assertTrue("Wrong value " + value.f1.lat, value.f1.lat == counter);
if (value.f1.lat % 2 == 0) {
assertNull("key was not null", value.f0);
} else {
Assert.assertTrue("Wrong value " + value.f0, value.f0 == counter);
}
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read KV from Kafka");
deleteTestTopic(topic);
}
示例2
/**
* Test delete behavior and metrics for producer.
* @throws Exception
*/
public void runAllDeletesTest() throws Exception {
final String topic = "alldeletestest";
createTestTopic(topic, 1, 1);
final int elementCount = 300;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<byte[], PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<byte[], PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<byte[], PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
final byte[] key = new byte[200];
rnd.nextBytes(key);
ctx.collect(new Tuple2<>(key, (PojoValue) null));
}
}
@Override
public void cancel() {
}
});
TypeInformationKeyValueSerializationSchema<byte[], PojoValue> schema = new TypeInformationKeyValueSerializationSchema<>(byte[].class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
producerProperties.putAll(secureProps);
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write deletes to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<byte[], PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, schema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<byte[], PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<byte[], PojoValue> value, Collector<Object> out) throws Exception {
// ensure that deleted messages are passed as nulls
assertNull(value.f1);
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read deletes from Kafka");
deleteTestTopic(topic);
}
示例3
public void runKeyValueTest() throws Exception {
final String topic = "keyvaluetest";
createTestTopic(topic, 1, 1);
final int elementCount = 5000;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<Long, PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<Long, PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<Long, PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
PojoValue pojo = new PojoValue();
pojo.when = new Date(rnd.nextLong());
pojo.lon = rnd.nextLong();
pojo.lat = i;
// make every second key null to ensure proper "null" serialization
Long key = (i % 2 == 0) ? null : i;
ctx.collect(new Tuple2<>(key, pojo));
}
}
@Override
public void cancel() {
}
});
KeyedSerializationSchema<Tuple2<Long, PojoValue>> schema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write KV to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
KafkaDeserializationSchema<Tuple2<Long, PojoValue>> readSchema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<Long, PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, readSchema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<Long, PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<Long, PojoValue> value, Collector<Object> out) throws Exception {
// the elements should be in order.
Assert.assertTrue("Wrong value " + value.f1.lat, value.f1.lat == counter);
if (value.f1.lat % 2 == 0) {
assertNull("key was not null", value.f0);
} else {
Assert.assertTrue("Wrong value " + value.f0, value.f0 == counter);
}
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read KV from Kafka");
deleteTestTopic(topic);
}
示例4
/**
* Test delete behavior and metrics for producer.
* @throws Exception
*/
public void runAllDeletesTest() throws Exception {
final String topic = "alldeletestest";
createTestTopic(topic, 1, 1);
final int elementCount = 300;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<byte[], PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<byte[], PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<byte[], PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
final byte[] key = new byte[200];
rnd.nextBytes(key);
ctx.collect(new Tuple2<>(key, (PojoValue) null));
}
}
@Override
public void cancel() {
}
});
TypeInformationKeyValueSerializationSchema<byte[], PojoValue> schema = new TypeInformationKeyValueSerializationSchema<>(byte[].class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
producerProperties.putAll(secureProps);
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write deletes to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<byte[], PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, schema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<byte[], PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<byte[], PojoValue> value, Collector<Object> out) throws Exception {
// ensure that deleted messages are passed as nulls
assertNull(value.f1);
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read deletes from Kafka");
deleteTestTopic(topic);
}
示例5
public void runKeyValueTest() throws Exception {
final String topic = "keyvaluetest";
createTestTopic(topic, 1, 1);
final int elementCount = 5000;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
DataStream<Tuple2<Long, PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<Long, PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<Long, PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
PojoValue pojo = new PojoValue();
pojo.when = new Date(rnd.nextLong());
pojo.lon = rnd.nextLong();
pojo.lat = i;
// make every second key null to ensure proper "null" serialization
Long key = (i % 2 == 0) ? null : i;
ctx.collect(new Tuple2<>(key, pojo));
}
}
@Override
public void cancel() {
}
});
KeyedSerializationSchema<Tuple2<Long, PojoValue>> schema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write KV to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
KafkaDeserializationSchema<Tuple2<Long, PojoValue>> readSchema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig());
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<Long, PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, readSchema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<Long, PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<Long, PojoValue> value, Collector<Object> out) throws Exception {
// the elements should be in order.
Assert.assertTrue("Wrong value " + value.f1.lat, value.f1.lat == counter);
if (value.f1.lat % 2 == 0) {
assertNull("key was not null", value.f0);
} else {
Assert.assertTrue("Wrong value " + value.f0, value.f0 == counter);
}
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read KV from Kafka");
deleteTestTopic(topic);
}
示例6
/**
* Test delete behavior and metrics for producer.
* @throws Exception
*/
public void runAllDeletesTest() throws Exception {
final String topic = "alldeletestest";
createTestTopic(topic, 1, 1);
final int elementCount = 300;
// ----------- Write some data into Kafka -------------------
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
DataStream<Tuple2<byte[], PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<byte[], PojoValue>>() {
@Override
public void run(SourceContext<Tuple2<byte[], PojoValue>> ctx) throws Exception {
Random rnd = new Random(1337);
for (long i = 0; i < elementCount; i++) {
final byte[] key = new byte[200];
rnd.nextBytes(key);
ctx.collect(new Tuple2<>(key, (PojoValue) null));
}
}
@Override
public void cancel() {
}
});
TypeInformationKeyValueSerializationSchema<byte[], PojoValue> schema = new TypeInformationKeyValueSerializationSchema<>(byte[].class, PojoValue.class, env.getConfig());
Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
producerProperties.setProperty("retries", "3");
producerProperties.putAll(secureProps);
kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
env.execute("Write deletes to Kafka");
// ----------- Read the data again -------------------
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
DataStream<Tuple2<byte[], PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, schema, props));
fromKafka.flatMap(new RichFlatMapFunction<Tuple2<byte[], PojoValue>, Object>() {
long counter = 0;
@Override
public void flatMap(Tuple2<byte[], PojoValue> value, Collector<Object> out) throws Exception {
// ensure that deleted messages are passed as nulls
assertNull(value.f1);
counter++;
if (counter == elementCount) {
// we got the right number of elements
throw new SuccessException();
}
}
});
tryExecute(env, "Read deletes from Kafka");
deleteTestTopic(topic);
}
