Use case is similar. But not able to check heap space issue, as data size
is small. Thought of mean while checking with you.
Thanks for looking into it. Really appreciate it.
I have marked the usage of temporal tables in bold red for ease of
reference.
On Fri, Sep 20, 2019, 8:18 PM Fabian Hueske <fhue...@gmail.com> wrote:
> Hi,
>
> This looks OK on the first sight.
> Is it doing what you expect?
>
> Fabian
>
> Am Fr., 20. Sept. 2019 um 16:29 Uhr schrieb Nishant Gupta <
> nishantgupta1...@gmail.com>:
>
>> Hi Fabian,
>>
>> Thanks for the information.
>> I have been reading about it and doing the same as a part of flink job
>> written in Java
>>
>> I am using proctime for both the tables. Can you please verify once the
>> implementation of temporal tables
>>
>>
>> here is the snippet.
>> ----------------------------
>> public class StreamingJob {
>>
>> public static void main(String[] args) throws Exception {
>>
>> ParameterTool params = ParameterTool.fromArgs(args);
>> final StreamExecutionEnvironment env =
>> StreamExecutionEnvironment.getExecutionEnvironment();
>> StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
>>
>> Properties kafkaConsumerProperties = new Properties();
>>
>> kafkaConsumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"X.X.X.X:9092");
>> kafkaConsumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
>> "cg54");
>> kafkaConsumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
>> "latest");
>>
>>
>> kafkaConsumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
>>
>> kafkaConsumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
>>
>> DataStream<String> badipStream = env.addSource(new
>> FlinkKafkaConsumer<>("badips", new SimpleStringSchema(),
>> kafkaConsumerProperties));
>>
>> DataStream<String> badipStreamM =
>> badipStream
>> .map(new MapFunction<String, String>() {
>> private static final long serialVersionUID = -686775202L;
>> @Override
>> public String map(String value) throws Exception {
>> try {
>> String[] v = value.split("\\t");
>> if(v.length > 1) {
>> return v[0].toString();
>> } else
>> return "0.0.0.0";
>> } catch (Exception e) {
>> System.err.println(e);
>> return "0.0.0.0";
>> }
>> }
>> });
>>
>> Table badipTable = tableEnv.fromDataStream(badipStreamM, *"bad_ip,
>> r_proctime.proctime");*
>>
>> tableEnv.registerTable("BadIP", badipTable);
>> TemporalTableFunction badIPTT =
>> badipTable.createTemporalTableFunction("r_proctime", "bad_ip");
>> tableEnv.registerFunction("BadIPTT", badIPTT);
>>
>>
>>
>> DataStream<ObjectNode> inKafkaStream = env
>> .addSource(new FlinkKafkaConsumer<>("tests", new
>> JSONKeyValueDeserializationSchema(false), kafkaConsumerProperties));
>> DataStream<Tuple2<String,String>> inKafkaStreamM =
>> inKafkaStream
>> .rebalance()
>> .filter(value -> value != null)
>> .map(new MapFunction<ObjectNode, Tuple2<String,String>>() {
>> private static final long serialVersionUID = -6867120202L;
>> @Override
>> public Tuple2<String,String> map(ObjectNode node) throws Exception
>> {
>> try {
>> ObjectNode nodeValue = (ObjectNode) node.get("value");
>> return new Tuple2<>(nodeValue.get("source.ip").asText(),
>> nodeValue.get("destination.ip").asText());
>> } catch (Exception e) {
>> System.err.println(e);
>> System.out.println(node);
>> return null;
>> }
>> }
>> });
>>
>> Table kafkaSource = tableEnv.fromDataStream(inKafkaStreamM, *"sourceIp,
>> destinationIp, k_proctime.proctime"*);
>> tableEnv.registerTable("KafkaSource", kafkaSource);
>> * Table resultKafkaMalicious = tableEnv.sqlQuery( "SELECT K.sourceIp,
>> K.destinationIp FROM KafkaSource AS K, LATERAL TABLE
>> (BadIPTT(K.k_proctime)) AS B WHERE K.sourceIp=B.bad_ip");*
>>
>> TupleTypeInfo<Tuple2<String, String>> tupleType = new TupleTypeInfo<>(
>> Types.STRING(),
>> Types.STRING());
>>
>> DataStream<Tuple2<String,String>> outStreamMalicious =
>> tableEnv.toAppendStream(resultKafkaMalicious, tupleType);
>>
>> Properties kafkaProducerProperties = new Properties();
>>
>> kafkaProducerProperties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"X.X.X.X:9092");
>> kafkaProducerProperties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
>> kafkaProducerProperties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");
>>
>> ObjectMapper mapper = new ObjectMapper();
>> DataStream<String> sinkStreamMaliciousData = outStreamMalicious
>> .map(new MapFunction<Tuple2<String,String>,String>() {
>> private static final long serialVersionUID = -6347120202L;
>> @Override
>> public String map(Tuple2<String,String> tuple) throws Exception {
>> try {
>> ObjectNode node = mapper.createObjectNode();
>> node.put("source.ip", tuple.f0);
>> node.put("destination.ip", tuple.f1);
>> return node.toString();
>> } catch (Exception e) {
>> System.err.println(e);
>> System.out.println(tuple);
>> return null;
>> }
>> }
>> });
>>
>>
>> sinkStreamMaliciousData.addSink(new
>> FlinkKafkaProducer<>("recon-data-malicious", new SimpleStringSchema(),
>> kafkaProducerProperties));
>> env.execute("Flink List Matching");
>> }
>> -------------------------------------------------------
>>
>> On Wed, Sep 18, 2019 at 6:09 PM Fabian Hueske <fhue...@gmail.com> wrote:
>>
>>> Hi Nishant,
>>>
>>> You should model the query as a join with a time-versioned table [1].
>>> The bad-ips table would be the time-time versioned table [2].
>>> Since it is a time-versioned table, it could even be updated with new
>>> IPs.
>>>
>>> This type of join will only keep the time-versioned table (the bad-ips
>>> in state) and not the other (high-volume) table.
>>>
>>> Best,
>>> Fabian
>>>
>>> [1]
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/joins.html#join-with-a-temporal-table
>>> [2]
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/temporal_tables.html
>>>
>>> Am Mi., 18. Sept. 2019 um 14:34 Uhr schrieb Nishant Gupta <
>>> nishantgupta1...@gmail.com>:
>>>
>>>> Hi Fabian,
>>>>
>>>> Thanks for your reply
>>>> I have a continuous stream of kafka coming and static table of badips.
>>>> I wanted to segregate records having bad ip.
>>>>
>>>> So therefore i was joining it. But with that 60 gb memory getting run
>>>> out
>>>>
>>>> So i used below query.
>>>> Can u please guide me in this regard
>>>>
>>>> On Wed, 18 Sep 2019 at 5:53 PM, Fabian Hueske <fhue...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> The query that you wrote is not a time-windowed join.
>>>>>
>>>>> INSERT INTO sourceKafkaMalicious
>>>>> SELECT sourceKafka.* FROM sourceKafka INNER JOIN badips ON
>>>>> sourceKafka.`source.ip`=badips.ip
>>>>> WHERE sourceKafka.timestamp_received BETWEEN CURRENT_TIMESTAMP -
>>>>> INTERVAL '15' MINUTE AND CURRENT_TIMESTAMP;
>>>>>
>>>>> The problem is the use of CURRENT_TIMESTAMP instead of a processing
>>>>> time (or event time) attribute of badips.
>>>>>
>>>>> What exactly are you trying to achieve with the query?
>>>>>
>>>>> Best, Fabian
>>>>>
>>>>> Am Mi., 18. Sept. 2019 um 14:02 Uhr schrieb Nishant Gupta <
>>>>> nishantgupta1...@gmail.com>:
>>>>>
>>>>>> Hi Team,
>>>>>>
>>>>>> I am running a query for Time Window Join as below
>>>>>>
>>>>>> INSERT INTO sourceKafkaMalicious
>>>>>> SELECT sourceKafka.* FROM sourceKafka INNER JOIN badips ON
>>>>>> sourceKafka.`source.ip`=badips.ip
>>>>>> WHERE sourceKafka.timestamp_received BETWEEN CURRENT_TIMESTAMP -
>>>>>> INTERVAL '15' MINUTE AND CURRENT_TIMESTAMP;
>>>>>>
>>>>>> Time windowed join, Flink SQL should automatically clear older
>>>>>> records, Some how the query does not clear the heapspace and fails
>>>>>> with error after sometime.
>>>>>>
>>>>>> Can you please let me know what could go wrong, or is it a issue
>>>>>>
>>>>>> Environment File chunks
>>>>>>
>>>>>> --------------------------------------------------------------------------------------------------------------------------------------------------------------
>>>>>> tables:
>>>>>> - name: sourceKafka
>>>>>> type: source-table
>>>>>> update-mode: append
>>>>>> connector:
>>>>>> type: kafka
>>>>>> version: "universal"
>>>>>> topic: test-data-flatten
>>>>>> properties:
>>>>>> - key: zookeeper.connect
>>>>>> value: x.x.x.x:2181
>>>>>> - key: bootstrap.servers
>>>>>> value: x.x.x.x:9092
>>>>>> - key: group.id
>>>>>> value: testgroup
>>>>>> format:
>>>>>> type: json
>>>>>> fail-on-missing-field: false
>>>>>> json-schema: >
>>>>>> {
>>>>>> type: 'object',
>>>>>> properties: {
>>>>>> 'source.ip': {
>>>>>> type: 'string'
>>>>>> },
>>>>>> 'source.port': {
>>>>>> type: 'string'
>>>>>> }
>>>>>> }
>>>>>> }
>>>>>> derive-schema: false
>>>>>> schema:
>>>>>> - name: ' source.ip '
>>>>>> type: VARCHAR
>>>>>> - name: 'source.port'
>>>>>> type: VARCHAR
>>>>>>
>>>>>> - name: sourceKafkaMalicious
>>>>>> type: sink-table
>>>>>> update-mode: append
>>>>>> connector:
>>>>>> type: kafka
>>>>>> version: "universal"
>>>>>> topic: test-data-mal
>>>>>> properties:
>>>>>> - key: zookeeper.connect
>>>>>> value: x.x.x.x:2181
>>>>>> - key: bootstrap.servers
>>>>>> value: x.x.x.x:9092
>>>>>> - key: group.id
>>>>>> value: testgroupmal
>>>>>> format:
>>>>>> type: json
>>>>>> fail-on-missing-field: false
>>>>>> json-schema: >
>>>>>> {
>>>>>> type: 'object',
>>>>>> properties: {
>>>>>> 'source.ip': {
>>>>>> type: 'string'
>>>>>> },
>>>>>> 'source.port': {
>>>>>> type: 'string'
>>>>>> }
>>>>>> }
>>>>>> }
>>>>>> derive-schema: false
>>>>>> schema:
>>>>>> - name: ' source.ip '
>>>>>> type: VARCHAR
>>>>>> - name: 'source.port'
>>>>>> type: VARCHAR
>>>>>>
>>>>>> - name: badips
>>>>>> type: source-table
>>>>>> #update-mode: append
>>>>>> connector:
>>>>>> type: filesystem
>>>>>> path: "/home/cyanadmin/ipsum/levels/badips.csv"
>>>>>> format:
>>>>>> type: csv
>>>>>> fields:
>>>>>> - name: ip
>>>>>> type: VARCHAR
>>>>>> comment-prefix: "#"
>>>>>> schema:
>>>>>> - name: ip
>>>>>> type: VARCHAR
>>>>>>
>>>>>> execution:
>>>>>> planner: blink
>>>>>> type: streaming
>>>>>> time-characteristic: event-time
>>>>>> periodic-watermarks-interval: 200
>>>>>> result-mode: table
>>>>>> max-table-result-rows: 1000000
>>>>>> parallelism: 3
>>>>>> max-parallelism: 128
>>>>>> min-idle-state-retention: 0
>>>>>> max-idle-state-retention: 0
>>>>>> restart-strategy:
>>>>>> type: fallback
>>>>>>
>>>>>> configuration:
>>>>>> table.optimizer.join-reorder-enabled: true
>>>>>> table.exec.spill-compression.enabled: true
>>>>>> table.exec.spill-compression.block-size: 128kb
>>>>>> Properties that describe the cluster to which table programs are
>>>>>> submitted to.
>>>>>>
>>>>>> deployment:
>>>>>> response-timeout: 5000
>>>>>>
>>>>>>
>>>>>> --------------------------------------------------------------------------------------------------------------------------------------------------------------
>>>>>>
>>>>>
