flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: window function not working when control stream broadcast
Date Mon, 13 Mar 2017 15:08:55 GMT
Hi Sam,

could you please also send the code of the timestamp/watermark assigner?
This could also affect things.


Best,

Aljoscha





On Thu, Mar 9, 2017, at 19:58, Sam Huang wrote:

> Hi Aljoscha,

> 

> Here's the code:

> private static class DataFilterFunImpl extends
> RichCoFlatMapFunction<KVTuple6, String, KVTuple6> {
>
>         private JSONParser parser; private Map<String, Map<String,
>         ControlJsonConfig>> whiteListMap = new HashMap<>();
>
>         @Override // tuple5(domain, device_type, type, key,
>         count_or_sum) public void flatMap1(KVTuple6 dataTuple,
>         Collector<KVTuple6> collector) throws Exception {
>
>             String type = dataTuple.f2; String[] keyValue = dataTuple-
>             .f3.split(RawEventExtractor.Constants.*DEFAULT_VALUE_SP*);
>             String key = keyValue[]; switch (type) {
>
>                 case RawEventExtractor.Constants.*VALUE_COUNT*: {
> 

>                     if (whiteListMap.containsKey(key)) {
> 

>                         ControlJsonConfig ruleConfig = whiteListMap.g-
>                         et(key).get(RawEventExtractor.Constants.*VALU-
>                         E_COUNT*); if (ruleConfig != null) {
>
>                             String value = keyValue.length > 1 ?
>                             keyValue[1] : ""; String bucket =
>                             ruleConfig.getBucketName(value); if
>                             (bucket != null) {
>
>                                 dataTuple.setField(String.*join*(RawE-
>                                 ventExtractor.Constants.*DEFAULT_VALU-
>                                 E_SP*, key, bucket), 3);
>                                 collector.collect(dataTuple); }
>
>                         } else {
> 

>                             collector.collect(dataTuple); }
>
>                     }
> 

>                     break;
>                 }
> 

>                 case RawEventExtractor.Constants.*VALUE_SUM*: {
> 

>                     if (whiteListMap.containsKey(key) && whiteListMap-
>                     .get(key).containsKey(RawEventExtractor.Constants-
>                     .*VALUE_SUM*)) {
>
>                         collector.collect(dataTuple); }
>
>                     break;
>                 }
> 

>                 default: collector.collect(dataTuple); }
>
>         }
> 

> 
> 

> 
> 

>         @Override public void flatMap2(String jsonStr,
>         Collector<KVTuple6> collector) throws Exception {
>
> //            Map<String, Map<String, ControlJsonConfig>> whiteListMap
> = whiteListMapState.value();            try {
>
>                 if (parser == null) {
> 

>                      parser = new JSONParser(); }
>
>                 JSONObject jsonConfig = (JSONObject)
>                 parser.parse(jsonStr); Tuple2<String, Map<String,
>                 ControlJsonConfig>> config =
>                 RawEventExtractor.*getKeyConfig*(jsonConfig); if
>                 (config.f1 == null) {
>
>                     whiteListMap.remove(config.f0); } else {
>
>                     whiteListMap.put(config.f0, config.f1); }
>
>             } catch (Exception e) {}
> 

>         }
> 

>     }
> 

> 

> FYI, if I setParallelism of both the control stream and data stream,
> the window function works. Is it necessary to do so for broadcast()
> function?
> 

> 

> On Thu, Mar 9, 2017 at 2:26 AM, Aljoscha Krettek
> <aljoscha@apache.org> wrote:
>> __

>> Hi Sam,

>> could you please also send the code for the DataFilterFunImpl and
>> your timestamps/watermark assigner. That could help in figuring out
>> the problem.
>> 

>> Best,

>> Aljoscha

>> 

>> 

>> On Wed, Mar 8, 2017, at 19:56, Sam Huang wrote:

>>> Hi Timo,

>>> 

>>> The window function sinks the data into InfluxDB, and it's not
>>> triggered.
>>> If I comment the ".timeWindow", and print results after the reduce
>>> function, it works
>>> Code for window function is here:

>>> 

>>> private static class WindowFunImpl implements
>>> WindowFunction<KVTuple6,Point,Tuple,TimeWindow> {
>>>
>>> 
>>> 
>>> 

>>>     @Override public void apply(Tuple tuple, TimeWindow window,
>>>     Iterable<KVTuple6> iterable,                  Collector<Point>
>>>     collector) throws Exception {
>>>
>>> 
>>> 
>>> 

>>>         KVTuple6 kvTypeTuple = iterable.iterator().next();
>>>         System.*out*.println("window: " + kvTypeTuple);
>>>         // Doesn't work here if use broadcast Point.Builder builder
>>>         = Point.*measurement*(*INFLUXDB_MEASUREMENT*)
>>>
>>> 
>>> 
>>> 

>>>                 .time(window.getStart(), TimeUnit.*MILLISECONDS*)
>>>
>>> 
>>> 
>>> 

>>>                 .tag(*TAG_DOMAIN*, kvTypeTuple.f0)
>>> 

>>> 
>>> 
>>> 

>>>                 .tag(*TAG_DEVICE*, kvTypeTuple.f1)
>>> 

>>> 
>>> 
>>> 

>>>                 .tag(*TAG_TYPE*, kvTypeTuple.f2)
>>> 

>>> 
>>> 
>>> 

>>>                 .tag(*TAG_KEY*, kvTypeTuple.f3)
>>> 

>>> 
>>> 
>>> 

>>>                 .addField(*FIELD*, kvTypeTuple.f4);
>>>
>>>         collector.collect(builder.build()); }
>>>
>>> 
>>> 
>>> 

>>> }
>>> 

>>> 
>>> 
>>> 

>>> 

>>> On Wed, Mar 8, 2017 at 1:10 AM, Timo Walther <twalthr@apache.org>
>>> wrote:
>>>> Hi Sam,

>>>> 

>>>> could you explain the behavior a bit more? How does the window
>>>> function behave? Is it not triggered or what is the content? What
>>>> is the result if you don't use a window function?
>>>> 

>>>> Timo

>>>> 

>>>> 

>>>> Am 08/03/17 um 02:59 schrieb Sam Huang:

>>>> 

>>>>> btw, the reduce function works well, I've printed out the data,
>>>>> and they are
>>>>> all correct. So are the timestamps and watermarks. And if I remove
>>>>> ".broadcast()", the data is successfully sinked.

>>>>> 

>>>>> Any help?

>>>>> 

>>>>> 

>>>>> 

>>>>> --

>>>>> View this message in context:
>>>>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/window-function-not-working-when-control-stream-broadcast-tp12093p12094.html
>>>>> Sent from the Apache Flink User Mailing List archive. mailing list
>>>>> archive at Nabble.com.
>>>> 

>> 



Mime
View raw message