kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Guozhang Wang (JIRA)" <j...@apache.org>
Subject [jira] [Created] (KAFKA-3902) Optimize KTable.filter() to reduce unnecessary traffic
Date Sat, 25 Jun 2016 05:05:16 GMT
Guozhang Wang created KAFKA-3902:
------------------------------------

             Summary: Optimize KTable.filter() to reduce unnecessary traffic
                 Key: KAFKA-3902
                 URL: https://issues.apache.org/jira/browse/KAFKA-3902
             Project: Kafka
          Issue Type: Bug
          Components: streams
            Reporter: Guozhang Wang


{{KTable.filter()}} operator is implemented in {{KTableFilter}}, and can be optimized to reduce
unnecessary data traffic to downstream operators. More specifically:

1. Some context: when a KTable participates in a downstream operators (e.g. if that operator
is an aggregation), then we need to materialize this KTable and send both its old value as
well as new value as a pair {old -> new} to the downstream operator. In practice it usually
needs to send the pair. 

So let's discuss about them separately, take the following example source stream for your
KTable

{{<a: 1>, <b: 2>, <a: 3> ...}}

When the KTable needs to be materialized, it will transform the source messages into the pairs
of:

{{<a: {null -> 1}>, <b: {nul -> 2}>, <a: {1 -> 3}>}}

2. If "send old value" is not enabled, then when the filter predicate returns false, we MUST
send a <key: null> to the downstream operator to indicate that this key is being filtered
in the table. Otherwise, for example if your filter is "value < 2", then the updated value
<a: 3> will just be filtered, resulting in incorrect semantics.

If it returns true we should still send the original <key: value> to downstream operators.

3. If "send old value" is enabled, then there are a couple of cases we can consider:

    a. If old value is <key: null> and new value is <key: not-null>, and the filter
predicate return false for the new value, then in this case it is safe to optimize and not
returning anything to the downstream operator, since in this case we know there is no value
for the key previously anyways; otherwise we send the original pair.

    b. If old value is <key: not-null> and new value is <key: null>, indicating
to delete this key, and the filter predicate return false for the old value, then in this
case it is safe to optimize and not returning anything to the downstream operator, since we
know that the old value has already been filtered in a previous message; otherwise we send
the original pair.

    c. If both old and new values are not null, and:

        1) predicate return true on both, send the original pair;
        2) predicate return false on both, we can optimize and do not send anything;
        3) predicate return true on old and false on new, send the key: {old -> null};
        4) predicate return false on old and true on new, send the key: {null -> new};



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message