Streaming Input and Output

Function Description

In DataFlow, streaming input means that a single UDF execution can obtain multiple batches of input data. Correspondingly, streaming output means that a single UDF execution can produce multiple batches of output data. For details, see How to Use.

To use dataflow.pyflow or dataflow.method to construct UDFs, streaming input is recommended if you need to batch multiple input data entries. Streaming output is recommended if you need to split output data into multiple parts for distribution.

Streaming input is enabled by setting stream_input='Queue'. In this case, the input parameter type is FlowMsgQueue, and users can fetch data from the queue on demand. Streaming output is implemented using Python yield generators. The number of output entries equals the number of yield statements executed.

Constraints

In streaming input scenarios, the DataFlow framework does not support data alignment or exception handling.

How to Use

Streaming input example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import dataflow as df

# Use the stream_input parameter to enable streaming input for the function. Inputs are queues. One execution of func fetches two data entries from input queue a and one entry from input queue b.
@df.pyflow(stream_input='Queue')
def func(a, b):
    data1 = a.get()
    data2 = a.get()
    data3 = b.get()
    return data1 + data2 + data3

@df.pyflow
class Foo():
    # Use the stream_input parameter to enable streaming input for the function. Inputs are queues. One execution of func fetches two data entries from input queue a and one entry from input queue b.
    @df.method(stream_input='Queue')
    def func(self, a, b):
        data1 = a.get()
        data2 = a.get()
        data3 = b.get()
        return data1 + data2 + data3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import dataflow as df

# Use yield to enable streaming output for the function. In this example, five outputs are generated per single scheduling run of func.
@df.pyflow
def func(a):
    for i in range(5):
        yield a + i


@df.pyflow
class Foo():
    # Use yield to enable streaming output for the function. In this example, five outputs are generated per single scheduling run of func.
    @df.method()
    def func(self, a):
        for i in range(5):
            yield a + i