Xiao-zhen-Liu commented on issue #4074:
URL: https://github.com/apache/texera/issues/4074#issuecomment-5900325986
@hhvu0102 Thanks for the follow-up questions.
**1. Using the data from both ports at once**
Xinyuan answered this in #8683: with the dependency from port 1 to port 0,
call your function in the port 1 branch. If you'd rather not rely on the order,
keep each table and call the function when both have arrived. `process_table`
is called once for each port, and only after all of that port's data has
arrived, so you don't need any code to check that the data has finished loading:
```python
from pytexera import *
class ProcessTableOperator(UDFTableOperator):
@overrides
def open(self) -> None:
self.tables = {}
@overrides
def process_table(self, table: Table, port: int) ->
Iterator[Optional[TableLike]]:
self.tables[port] = table
if len(self.tables) == 2:
# both ports have all their data; this runs once
yield your_function(self.tables[0], self.tables[1])
```
Both ways assume "Worker count" is 1, which is the default. With more
workers, each port's data is split among the workers, and each worker would
call `your_function` with only its part of the data.
**2. The wiki example**
You're right: the example only works because port 0 finishes before port 1,
and the wiki doesn't point that out. It is written for the "2-in Python UDF"
operator, where port 1 (`tuples`) always waits until port 0 (`model`) has
finished, so `self.model` is set before any tuple arrives on port 1. In a
regular Python UDF with two ports there is no such wait unless you add the
dependency, so the example fails when port 1's data comes first. It also fails
if port 0 gets no tuples at all. The Python UDF guide is being rewritten in
#8737, and the new version says that inputs arrive in no fixed order and shows
the pattern above.
**3. Why it works this way**
Texera uses pipelined execution: operators run at the same time, and data is
passed to the next operator while it is being produced, not after the previous
operator finishes. This has three benefits: results start to appear sooner,
operators work in parallel instead of waiting for each other, and intermediate
results don't have to be stored in full between operators. To make this work,
each operator, built-in or UDF, receives its input one tuple at a time,
together with the port the tuple came in on. That is the Tuple API
(`process_tuple`). The Table API is built on top of it: it collects each port's
tuples and calls `process_table` when that port has no more data, so it is
called once per port.
For the same reasons, the engine doesn't wait for all ports by default: an
operator can start on one input while the operators feeding its other input are
still running. When an operator does need one input to be complete before
another, it declares a dependency between the two ports, and the engine
finishes the first port before it sends any data to the second. The built-in
Hash Join works this way: it builds a hash table from the whole left input,
then looks up each right tuple in it. The 2-in Python UDF has this dependency
built in, and in the regular Python UDF you can add it yourself, as in the GIF
above. Without a dependency, the order depends on which upstream operator
finishes first, which is why port 1 came first in your workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
