[SPARK-24396] [SS] [PYSPARK] Add Structured Streaming ForeachWriter for python #21477
Conversation
dev/sparktestsupport/modules.py
Outdated
| @@ -389,19 +389,7 @@ def __hash__(self): | |||
| "python/pyspark/sql" | |||
| ], | |||
| python_test_goals=[ | |||
| "pyspark.sql.types", | |||
There was a problem hiding this comment.
This is only to speed up local testing. Will remove this.
python/pyspark/sql/tests.py
Outdated
| @@ -296,6 +296,7 @@ def tearDown(self): | |||
| # tear down test_bucketed_write state | |||
| self.spark.sql("DROP TABLE IF EXISTS pyspark_bucket") | |||
|
|
|||
| ''' | |||
There was a problem hiding this comment.
This is only to speed up local testing. Will remove this.
tdas
changed the title
|
Test build #91371 has finished for PR 21477 at commit
|
|
jenkins retest this please. |
|
Test build #91385 has finished for PR 21477 at commit
|
|
Test build #91386 has finished for PR 21477 at commit
|
|
Test build #91384 has finished for PR 21477 at commit
|
|
Test build #91388 has finished for PR 21477 at commit
|
python/pyspark/sql/streaming.py
Outdated
| @@ -30,6 +30,7 @@ | |||
| from pyspark.sql.readwriter import OptionUtils, to_str | |||
| from pyspark.sql.types import * | |||
| from pyspark.sql.utils import StreamingQueryException | |||
| from abc import ABCMeta, abstractmethod | |||
python/pyspark/sql/streaming.py
Outdated
| """ | ||
| The provided object is a callable function that is supposed to be called on each row. | ||
| Construct a function that takes an iterator and calls the provided function on each row. | ||
| """ |
There was a problem hiding this comment.
I believe we should better turn this to comment form..
# The ..
# Construct ...
python/pyspark/sql/streaming.py
Outdated
| The provided object is not a callable function. Then it is expected to have a | ||
| 'process(row)' method, and optional 'open(partition_id, epoch_id)' and | ||
| 'close(error)' methods. | ||
| """ |
python/pyspark/sql/streaming.py
Outdated
| else: | ||
| open_exists = True | ||
|
|
||
| close_exists = False |
There was a problem hiding this comment.
maybe
close_exists = hasattr(f, "close")
if close_exists:
...
python/pyspark/sql/streaming.py
Outdated
| if not callable(getattr(f, 'process')): | ||
| raise Exception("Attribute 'process' in provided object is not callable") | ||
|
|
||
| open_exists = False |
python/pyspark/sql/tests.py
Outdated
| import random | ||
| file = open(os.path.join(dir, str(random.randint(0, 100000))), 'w') | ||
| file.write("%s\n" % str(event)) | ||
| file.close() |
There was a problem hiding this comment.
should it be put in finally too?
python/pyspark/sql/tests.py
Outdated
|
|
||
| def _write_event(self, dir, event): | ||
| import random | ||
| file = open(os.path.join(dir, str(random.randint(0, 100000))), 'w') |
There was a problem hiding this comment.
nit: file shadows the builtin function file ..
There was a problem hiding this comment.
We might feel more convenient with with statement, and renaming file to f or fw or so. Please ignore if there's specific reason not to use with statement.
There was a problem hiding this comment.
Why not use a uuid in the file name?
python/pyspark/sql/tests.py
Outdated
| sq.processAllAvailable() | ||
| self.fail("bad writer should fail the query") # this is not expected | ||
| except StreamingQueryException as e: | ||
| # self.assertTrue("test error" in e.desc) # this is expected |
There was a problem hiding this comment.
Eh, shall we uncomment this?
There was a problem hiding this comment.
This cannot be done because somehow e.desc does not have the original underlying error when it's a StreamingQueryException. I think this is a separate bug that needs to be fixed independent of this PR.
python/pyspark/sql/tests.py
Outdated
| self.assertEqual(len(tester.process_events()), 0) # no row was processed | ||
| close_events = tester.close_events() | ||
| self.assertEqual(len(close_events), 1) | ||
| # self.assertTrue("test error" in e[0]['error']) |
| } | ||
|
|
||
| def add(row: UnsafeRow): Unit = withLock { | ||
| assert(queue.add(row), s"Failed to add row to HybridRowQueue while sending data to Python" + |
|
let me cc @HeartSaVioR too since he's looking through ss code bit. |
|
Thanks @HyukjinKwon for cc-ing me. I didn't cover the python part on structured streaming so would take some time to cover and going through the code. Hoping I can participate reviewing in time. |
python/pyspark/sql/streaming.py
Outdated
| processing one partition of the data generated in a distributed manner. | ||
|
|
||
| * This object must be serializable because each task will get a fresh | ||
| serialized-deserializedcopy of the provided object. Hence, it is strongly |
There was a problem hiding this comment.
nit: deserialized copy (space)
python/pyspark/sql/streaming.py
Outdated
|
|
||
| * This object must be serializable because each task will get a fresh | ||
| serialized-deserializedcopy of the provided object. Hence, it is strongly | ||
| recommended that any initialization for writing data (e.g. opening a |
There was a problem hiding this comment.
any initialization for writing data (e.g. opening a connection or starting a transaction) be done open after the
open(...)method has been called
be done open seems a bit odd. If we can polish the sentence it would be better.
python/pyspark/sql/streaming.py
Outdated
| if not callable(getattr(f, 'process')): | ||
| raise Exception("Attribute 'process' in provided object is not callable") | ||
|
|
||
| open_exists = False |
There was a problem hiding this comment.
We might be able to extract the code block to the function, since the logic for checking open and close are exactly same.
python/pyspark/sql/streaming.py
Outdated
| for x in iterator: | ||
| f.process(x) | ||
| except Exception as ex: | ||
| call_close_if_needed(ex) |
There was a problem hiding this comment.
putting this to finally sounds like better pattern for me too, and then we may be able to inline call_close_if_needed.
python/pyspark/sql/tests.py
Outdated
|
|
||
| def _write_event(self, dir, event): | ||
| import random | ||
| file = open(os.path.join(dir, str(random.randint(0, 100000))), 'w') |
There was a problem hiding this comment.
We might feel more convenient with with statement, and renaming file to f or fw or so. Please ignore if there's specific reason not to use with statement.
python/pyspark/sql/tests.py
Outdated
| self.assertEqual(len(close_events), 1) | ||
| # self.assertTrue("test error" in e[0]['error']) | ||
|
|
||
| ''' |
There was a problem hiding this comment.
Leaving marker as well: This is only to speed up local testing. Will remove this.
There was a problem hiding this comment.
Yes i will after I finish adding all the tests.
| * usually should do all the initialization (e.g. opening a connection or initiating a transaction) | ||
| * in the `open` method. | ||
| * The abstract class for writing custom logic to process data generated by a query. | ||
| * This is often used to write the output of a streaming query to arbitrary storage systems. |
There was a problem hiding this comment.
Looks like doc is duplicated between foreach() and ForeachWriter. I'm not sure how we can leave some reference on Python doc instead of duplicating content, but even Python doc doesn't support some kind of reference, some part of content seems to be OK to be placed to either place, not both.
There was a problem hiding this comment.
I don't think there's a easy way to reference doc across other languages. It's unfortunate but usually we have duplicated the docs between other language API docs ...
There was a problem hiding this comment.
Ah yes my bad. I confused this as python.
| object PythonForeachWriter { | ||
|
|
||
| /** | ||
| * A buffer that is designed for the sole purpose of buffering UnsafeRows in PythonForeahWriter. |
There was a problem hiding this comment.
nit: PythonForeachWriter
| some input data. That would require you to specify the processing logic in the next | ||
| way. | ||
|
|
||
| #. An **object** with a ``process`` method and optional ``open`` and ``close`` methods. |
There was a problem hiding this comment.
@tdas, wouldn't we better just have ForeachWriter class?
There was a problem hiding this comment.
I discussed this with @marmbrus . If there is a ForeachWriter class in python, then uses will have to additionally import it. That's just another overhead that can be avoided by just allowing any class with the appropriate methods. One less step for python users.
| This is often used to write the output of a streaming query to arbitrary storage systems. | ||
| The processing logic can be specified in two ways. | ||
|
|
||
| #. A **function** that takes a row as input. |
There was a problem hiding this comment.
Seems this variant is specific to Python. I thought we should better match how we support with Scala side.
There was a problem hiding this comment.
This is superset of what we support in scala. Python users are more likely to use simple lambdas instead of defining classes. But they may also want to write transactional stuff in python with open and close methods. Hence providing both alternatives seems to be a good idea.
There was a problem hiding this comment.
(including the response to #21477 (comment)) I kind of agree that it's a-okay idea but I think we usually provide a consistent API support so far unless it's language specific, for example, ContextManager, decorator in Python and etc.
Just for clarification, does Scala side support function only support too?
Also, I know attribute-checking way is kind of more like "Pythonic" way but I am seeing the documentation is already diverted between Scala vs Python. It costs maintaining overhead on the other hand.
There was a problem hiding this comment.
I mean, we could maybe consider the other ways but wouldn't it better to have the consistent support as the primary, and then see if the other ways are really requested by users? I think we could still incrementally add attribute-checking way or the lambda (or function to be more correct) way later (but we can't in the opposite way).
There was a problem hiding this comment.
Python APIs anyways have slightly divergences from Scala/Java APIs in order to provide better experiences for Python users. For example, StreamingQuery.lastProgress returns an object of type StreamingQueryProgress in Java/Scala but returns a dict in python. Because python users are more used to dealing with dicts, and java/scala users (typed language) are more comfortable with structures). Similarly, in DataFrame.select, you can refer to columns in scala using $"columnName" but in python, you can refer to it as df.columnName. If we strictly adhere to pure consistency, then we cannot make it convenient for users in different languages. And ultimately convenience is what matters for the user experience. So its okay to have a superset of supported types in python compared to java/scala.
Personally, I think we should also add the lambda variant to Scala as well. But that decision for Scala is independent of this PR as there is enough justification for add the lambda variant for
There was a problem hiding this comment.
I believe $"columnName" is more like a language specific feature in Scala and I think df.columnName is language specific to Python.
And ultimately convenience is what matters for the user experience.
Thing is, it sounded to me like we are kind of prejudging it.. We can't revert it back easily once we go in this way ..
I think we should also add the lambda variant to Scala as well.
+1
I am okay but I hope this shouldn't be usually done next time ..
|
Seems fine so far to me otherwise |
python/pyspark/sql/streaming.py
Outdated
| open_exists = True | ||
|
|
||
| close_exists = False | ||
| if hasattr(f, "close"): |
There was a problem hiding this comment.
Also, for perfectness, we should check argument specification too ... although it's a bit too much. I felt like I had to mention it at least.
There was a problem hiding this comment.
It's all dynamically typed, right? I don't think there is a good way to check where the function accepts the right argument types without actually calling the function (please correct me if I am wrong). And this is a standard problem in python that is everybody using python is used to (i.e. runtime exceptions when using incorrectly typed parameters). Also, no other operations that take lambdas check the types and counts. So I think we should just be consistent and let that be.
There was a problem hiding this comment.
We can check argspec at least I believe. but, yup, I am fine with it as is.
python/pyspark/sql/streaming.py
Outdated
| mode, then this guarantee does not hold and therefore should not be used for | ||
| deduplication. | ||
|
|
||
| * The ``close()`` method (if exists) is will be called if `open()` method exists and |
| * @return `true` if the corresponding partition and version id should be processed. `false` | ||
| * indicates the partition should be skipped. | ||
| */ | ||
| def open(partitionId: Long, version: Long): Boolean | ||
| def open(partitionId: Long, epochId: Long): Boolean |
There was a problem hiding this comment.
Renaming a parameter breaks Scala source compatibility. I'm totally fine to change this since it's not a stable API, just point this out.
There was a problem hiding this comment.
oh right... it probably breaks source compatibility. I dont think that is fine :( especially in the 2.x line, even though it is marked as experimental (its been out there for over 2 years now, this is hardly experimental and we should not remove the experimental tag, but that's a different discussion).
There was a problem hiding this comment.
okay. I checked but there is no compatibility issue in this case. The source compatibility issue arises when a code calls a method with param name (e.g. func(paramName=value)) and the param name changes. In this case, users are overriding the method (overriding does not care about the exact name, only the types) and Spark is internally calling the method (not by param name). So this is fine.
python/pyspark/sql/tests.py
Outdated
|
|
||
| def _write_event(self, dir, event): | ||
| import random | ||
| file = open(os.path.join(dir, str(random.randint(0, 100000))), 'w') |
There was a problem hiding this comment.
Why not use a uuid in the file name?
|
Test build #91539 has finished for PR 21477 at commit
|
tdas
changed the title
|
LGTM |
|
Test build #91871 has finished for PR 21477 at commit
|
|
|
||
| if (count > 0) { | ||
| val row = queue.remove() | ||
| assert(row != null, s"HybridRowQueue.remove() returned null " + |
python/pyspark/sql/streaming.py
Outdated
| if should_process: | ||
| for x in iterator: | ||
| f.process(x) | ||
|
|
There was a problem hiding this comment.
I usually write try-except-finally without a newline tho ..
There was a problem hiding this comment.
This looks like strictly related to preference, and I prefer newline. Are you referring to the Spark style guide or PEP 8 or so?
There was a problem hiding this comment.
I see no newline between them is more consistent in the current codebase. It should be better to keep consistent.
There was a problem hiding this comment.
Oh OK. I agree it might be better to try best to keep consistent though style guide doesn't define it explicitly.
python/pyspark/sql/tests.py
Outdated
|
|
||
| self.assertEqual(len(tester.open_events()), 2) | ||
|
|
||
| self.assertEqual(len(tester.process_events()), 0) # no row was processed |
There was a problem hiding this comment.
Is that a PEP 8 rule or Spark style?
There was a problem hiding this comment.
PEP 8 claims (at least) 2 spaces before inlined comments (https://www.python.org/dev/peps/pep-0008/#inline-comments). So, the previous style is totally fine but I have been doing this with less spaces since I find there's no point of adding more spaces unless it looks more pretty or has a format.
(This was just a tiny nit too me. I am okay with just ignoring such my comments personally)
python/pyspark/sql/tests.py
Outdated
| tester.write_close_event(error) | ||
|
|
||
| tester.run_streaming_query_on_writer(ForeachWriter(), 2) | ||
| self.assertEqual(len(tester.open_events()), 0) # no open events |
|
(Please ignore nits. They are ignorable but I just left them while reading the codes in case you might want to address them) |
|
LGTM too just for clarification. |
|
Test build #91898 has finished for PR 21477 at commit
|
|
Thank you every one. I merging this. |
… python This PR adds `foreach` for streaming queries in Python. Users will be able to specify their processing logic in two different ways. - As a function that takes a row as input. - As an object that has methods `open`, `process`, and `close` methods. See the python docs in this PR for more details. Added java and python unit tests Author: Tathagata Das <tathagata.das1565@gmail.com> Closes apache#21477 from tdas/SPARK-24396. Ref: LIHADOOP-48531
What changes were proposed in this pull request?
This PR adds
foreachfor streaming queries in Python. Users will be able to specify their processing logic in two different ways.open,process, andclosemethods.See the python docs in this PR for more details.
How was this patch tested?
Added java and python unit tests