Streaming join performance

[Сыроватский Артем Иванович <a....@bftcom.com>]

Hello, Flink community!

I have some important use case for me, which shows extremely bad performance:

  *   Streaming application
  *   sql table api
  *   10 normal joins (state should be kept forever)

Join rules are simple, i have 10 ten tables, which have same primary key. I want to join result table from 10 pieces.

But Flink joins sequentionally, so i have a chain with 10 joins.

What happens if i generate update message for first table in chain:


  *   first join operator will produce 2 records: delete+insert
  *   second operator will double incoming messages. result=2*2=4 messages
  *   ...
  *   last operator will produce 2**10=1024 messages.

Perfomance become extremely slow and resources are wasting away.

I've made some simple compaction operator, which compacts records after join:


  *   join operator after receiving delete message, generates 2 messages
  *   after receiving insert message, generate 2 more messges
  *   but two of the four are compacted. So operator receives 2 messages->sends 2 messages

I wonder if this approach is right? Why it is not implemented in Flink yet?

And i've got some problem how should i implement it on cluster, because i have changed some flink sources, which are not pluggable?

I have modified StreamExecJoin class and added this code as a proof of concept:


final OneInputTransformation<RowData, RowData> compactTransform =
        ExecNodeUtil.createOneInputTransformation(
                transform,
                "compact join results",
                "description",
                new ProcessOperator<>(new CompactStreamOperator(equaliser)),
                InternalTypeInfo.of(returnType),
                leftTransform.getParallelism()
        );

return compactTransform;

Transform operator:
@Override

public void processElement(
        RowData value,
        ProcessFunction<RowData, RowData>.Context ctx,
        Collector<RowData> collector) throws Exception {

    counter++;

    boolean compacted=false;
    if (value.getRowKind()==RowKind.DELETE) {
        value.setRowKind(RowKind.INSERT);
        for (int i = buffer.size() - 1; i >= 0; i--) {
            RowData x = buffer.get(i);
            if (x.getRowKind() == RowKind.INSERT && recordEqualiser.equals(x, value)) {
                buffer.remove(i);
                compacted = true;
                break;
            }
        }
        value.setRowKind(RowKind.DELETE);
    }

    if (!compacted) {
        buffer.add(value);
    }

    if (counter>=10)
    {
        buffer.forEach(collector::collect);
        buffer.clear();
        counter=0;
    }
}


[cid:f886301c-4708-494e-a6df-d81137150774]



Regards,
Artem

Re: Streaming join performance

[liu ron <ro...@gmail.com>]

Hi, Arterm

The current Flink for multi-stream Join does have the problem of record
amplification, which is a pain point problem. For this problem, the
community is considering the idea of Mini-batch join, through the way of
buffer bundle record, then folding the record by RowKind, so as to reduce
the amplification problem. We have verified through the PoC gain is very
obvious. I think your overall idea is right, we need to do this
optimization from the framework level, and it can be controlled by options.

Best,
Ron



Сыроватский Артем Иванович <a....@bftcom.com> 于2023年8月4日周五 16:23写道：

> Hello, Flink community!
>
> I have some important use case for me, which shows extremely bad
> performance:
>
>    - Streaming application
>    - sql table api
>    - 10 normal joins (state should be kept forever)
>
>
> Join rules are simple, i have 10 ten tables, which have same primary key.
> I want to join result table from 10 pieces.
>
> But Flink joins sequentionally, so i have a chain with 10 joins.
>
> What happens if i generate update message for first table in chain:
>
>
>    - first join operator will produce 2 records: delete+insert
>    - second operator will double incoming messages. result=2*2=4 messages
>    - ...
>    - last operator will produce 2**10=1024 messages.
>
>
> Perfomance become extremely slow and resources are wasting away.
>
> I've made some simple compaction operator, which compacts records after
> join:
>
>
>    - join operator after receiving delete message, generates 2 messages
>    - after receiving insert message, generate 2 more messges
>    - but two of the four are compacted. So operator receives 2
>    messages->sends 2 messages
>
>
> I wonder if this approach is right? Why it is not implemented in Flink yet?
>
> And i've got some problem how should i implement it on cluster, because i
> have changed some flink sources, which are not pluggable?
>
> I have modified StreamExecJoin class and added this code as a proof of
> concept:
>
> final OneInputTransformation<RowData, RowData> compactTransform =
>         ExecNodeUtil.createOneInputTransformation(
>                 transform,
>                 "compact join results",
>                 "description",
>                 new ProcessOperator<>(new CompactStreamOperator(equaliser)),
>                 InternalTypeInfo.of(returnType),
>                 leftTransform.getParallelism()
>         );
>
> return compactTransform;
>
>
> Transform operator:
> @Override
>
> public void processElement(
>         RowData value,
>         ProcessFunction<RowData, RowData>.Context ctx,
>         Collector<RowData> collector) throws Exception {
>
>     counter++;
>
>     boolean compacted=false;
>     if (value.getRowKind()==RowKind.DELETE) {
>         value.setRowKind(RowKind.INSERT);
>         for (int i = buffer.size() - 1; i >= 0; i--) {
>             RowData x = buffer.get(i);
>             if (x.getRowKind() == RowKind.INSERT && recordEqualiser.equals(x, value)) {
>                 buffer.remove(i);
>                 compacted = true;
>                 break;
>             }
>         }
>         value.setRowKind(RowKind.DELETE);
>     }
>
>     if (!compacted) {
>         buffer.add(value);
>     }
>
>     if (counter>=10)
>     {
>         buffer.forEach(collector::collect);
>         buffer.clear();
>         counter=0;
>     }
> }
>
>
>
>
>
>
>
> Regards,
> Artem
>
>
>