class_docstr: Interface used to write a streaming :class:`DataFrame <pyspark.sql.DataFrame>` to external
  storage systems (e.g. file systems, key-value stores, etc).
  Use :attr:`DataFrame.writeStream <pyspark.sql.DataFrame.writeStream>`
  to access this.
  
  .. versionadded:: 2.0.0
  
  .. versionchanged:: 3.5.0
      Supports Spark Connect.
  
  Notes
  -----
  This API is evolving.
  
  Examples
  --------
  The example below uses Rate source that generates rows continuously.
  After that, we operate a modulo by 3, and then writes the stream out to the console.
  The streaming query stops in 3 seconds.
  
  >>> import time
  >>> df = spark.readStream.format("rate").load()
  >>> df = df.selectExpr("value % 3 as v")
  >>> q = df.writeStream.format("console").start()
  >>> time.sleep(3)
  >>> q.stop()
class_docstr_tok: ["Interface","used","to","write","a","streaming",":","class",":","`","DataFrame","<","pyspark.sql.DataFrame",">","`","to","external","storage","systems","(","e.g",".","file","systems",",","key-value","stores",",","etc",")",".","Use",":","attr",":","`","DataFrame.writeStream","<","pyspark.sql.DataFrame.writeStream",">","`","to","access","this",".","..","versionadded",":",":","2.0.0","..","versionchanged",":",":","3.5.0","Supports","Spark","Connect",".","Notes","--","--","-","This","API","is","evolving",".","Examples","--","--","--","--","The","example","below","uses","Rate","source","that","generates","rows","continuously",".","After","that",",","we","operate","a","modulo","by","3",",","and","then","writes","the","stream","out","to","the","console",".","The","streaming","query","stops","in","3","seconds",".",">",">",">","import","time",">",">",">","df","=","spark.readStream.format","(","``","rate","''",")",".load","(",")",">",">",">","df","=","df.selectExpr","(","``","value","%","3","as","v","''",")",">",">",">","q","=","df.writeStream.format","(","``","console","''",")",".start","(",")",">",">",">","time.sleep","(","3",")",">",">",">","q.stop","(",")"]
docstr: Sets the output of the streaming query to be processed using the provided writer ``f``.
  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.
      This is a simple way to express your processing logic. Note that this does
      not allow you to deduplicate generated data when failures cause reprocessing of
      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.
      The object can have the following methods.
  
      * ``open(partition_id, epoch_id)``: *Optional* method that initializes the processing
          (for example, open a connection, start a transaction, etc). Additionally, you can
          use the `partition_id` and `epoch_id` to deduplicate regenerated data
          (discussed later).
  
      * ``process(row)``: *Non-optional* method that processes each :class:`Row`.
  
      * ``close(error)``: *Optional* method that finalizes and cleans up (for example,
          close connection, commit transaction, etc.) after all rows have been processed.
  
      The object will be used by Spark in the following way.
  
      * A single copy of this object is responsible of all the data generated by a
          single task in a query. In other words, one instance is responsible for
          processing one partition of the data generated in a distributed manner.
  
      * This object must be serializable because each task will get a fresh
          serialized-deserialized copy of the provided object. Hence, it is strongly
          recommended that any initialization for writing data (e.g. opening a
          connection or starting a transaction) is done after the `open(...)`
          method has been called, which signifies that the task is ready to generate data.
  
      * The lifecycle of the methods are as follows.
  
          For each partition with ``partition_id``:
  
          ... For each batch/epoch of streaming data with ``epoch_id``:
  
          ....... Method ``open(partitionId, epochId)`` is called.
  
          ....... If ``open(...)`` returns true, for each row in the partition and
                  batch/epoch, method ``process(row)`` is called.
  
          ....... Method ``close(errorOrNull)`` is called with error (if any) seen while
                  processing rows.
  
      Important points to note:
  
      * The `partitionId` and `epochId` can be used to deduplicate generated data when
          failures cause reprocessing of some input data. This depends on the execution
          mode of the query. If the streaming query is being executed in the micro-batch
          mode, then every partition represented by a unique tuple (partition_id, epoch_id)
          is guaranteed to have the same data. Hence, (partition_id, epoch_id) can be used
          to deduplicate and/or transactionally commit data and achieve exactly-once
          guarantees. However, if the streaming query is being executed in the continuous
          mode, then this guarantee does not hold and therefore should not be used for
          deduplication.
  
      * The ``close()`` method (if exists) will be called if `open()` method exists and
          returns successfully (irrespective of the return value), except if the Python
          crashes in the middle.
  
  .. versionadded:: 2.4.0
  
  .. versionchanged:: 3.5.0
      Supports Spark Connect.
  
  Notes
  -----
  This API is evolving.
  
  Examples
  --------
  >>> import time
  >>> df = spark.readStream.format("rate").load()
  
  Print every row using a function
  
  >>> def print_row(row):
  ...     print(row)
  ...
  >>> q = df.writeStream.foreach(print_row).start()
  >>> time.sleep(3)
  >>> q.stop()
  
  Print every row using a object with process() method
  
  >>> class RowPrinter:
  ...     def open(self, partition_id, epoch_id):
  ...         print("Opened %d, %d" % (partition_id, epoch_id))
  ...         return True
  ...
  ...     def process(self, row):
  ...         print(row)
  ...
  ...     def close(self, error):
  ...         print("Closed with error: %s" % str(error))
  ...
  >>> q = df.writeStream.foreach(print_row).start()
  >>> time.sleep(3)
  >>> q.stop()
docstr_tok: ["Sets","the","output","of","the","streaming","query","to","be","processed","using","the","provided","writer","``","f","``",".","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",".","This","is","a","simple","way","to","express","your","processing","logic",".","Note","that","this","does","not","allow","you","to","deduplicate","generated","data","when","failures","cause","reprocessing","of","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",".","The","object","can","have","the","following","methods",".","*","``","open","(","partition_id",",","epoch_id",")","``",":","*","Optional","*","method","that","initializes","the","processing","(","for","example",",","open","a","connection",",","start","a","transaction",",","etc",")",".","Additionally",",","you","can","use","the","`","partition_id","`","and","`","epoch_id","`","to","deduplicate","regenerated","data","(","discussed","later",")",".","*","``","process","(","row",")","``",":","*","Non-optional","*","method","that","processes","each",":","class",":","`","Row","`",".","*","``","close","(","error",")","``",":","*","Optional","*","method","that","finalizes","and","cleans","up","(","for","example",",","close","connection",",","commit","transaction",",","etc",".",")","after","all","rows","have","been","processed",".","The","object","will","be","used","by","Spark","in","the","following","way",".","*","A","single","copy","of","this","object","is","responsible","of","all","the","data","generated","by","a","single","task","in","a","query",".","In","other","words",",","one","instance","is","responsible","for","processing","one","partition","of","the","data","generated","in","a","distributed","manner",".","*","This","object","must","be","serializable","because","each","task","will","get","a","fresh","serialized-deserialized","copy","of","the","provided","object",".","Hence",",","it","is","strongly","recommended","that","any","initialization","for","writing","data","(","e.g",".","opening","a","connection","or","starting","a","transaction",")","is","done","after","the","`","open","(","...",")","`","method","has","been","called",",","which","signifies","that","the","task","is","ready","to","generate","data",".","*","The","lifecycle","of","the","methods","are","as","follows",".","For","each","partition","with","``","partition_id","``",":","...","For","each","batch\/epoch","of","streaming","data","with","``","epoch_id","``",":",".......","Method","``","open","(","partitionId",",","epochId",")","``","is","called",".",".......","If","``","open","(","...",")","``","returns","true",",","for","each","row","in","the","partition","and","batch\/epoch",",","method","``","process","(","row",")","``","is","called",".",".......","Method","``","close","(","errorOrNull",")","``","is","called","with","error","(","if","any",")","seen","while","processing","rows",".","Important","points","to","note",":","*","The","`","partitionId","`","and","`","epochId","`","can","be","used","to","deduplicate","generated","data","when","failures","cause","reprocessing","of","some","input","data",".","This","depends","on","the","execution","mode","of","the","query",".","If","the","streaming","query","is","being","executed","in","the","micro-batch","mode",",","then","every","partition","represented","by","a","unique","tuple","(","partition_id",",","epoch_id",")","is","guaranteed","to","have","the","same","data",".","Hence",",","(","partition_id",",","epoch_id",")","can","be","used","to","deduplicate","and\/or","transactionally","commit","data","and","achieve","exactly-once","guarantees",".","However",",","if","the","streaming","query","is","being","executed","in","the","continuous","mode",",","then","this","guarantee","does","not","hold","and","therefore","should","not","be","used","for","deduplication",".","*","The","``","close","(",")","``","method","(","if","exists",")","will","be","called","if","`","open","(",")","`","method","exists","and","returns","successfully","(","irrespective","of","the","return","value",")",",","except","if","the","Python","crashes","in","the","middle",".","..","versionadded",":",":","2.4.0","..","versionchanged",":",":","3.5.0","Supports","Spark","Connect",".","Notes","--","--","-","This","API","is","evolving",".","Examples","--","--","--","--",">",">",">","import","time",">",">",">","df","=","spark.readStream.format","(","``","rate","''",")",".load","(",")","Print","every","row","using","a","function",">",">",">","def","print_row","(","row",")",":","...","print","(","row",")","...",">",">",">","q","=","df.writeStream.foreach","(","print_row",")",".start","(",")",">",">",">","time.sleep","(","3",")",">",">",">","q.stop","(",")","Print","every","row","using","a","object","with","process","(",")","method",">",">",">","class","RowPrinter",":","...","def","open","(","self",",","partition_id",",","epoch_id",")",":","...","print","(","``","Opened","%","d",",","%","d","''","%","(","partition_id",",","epoch_id",")",")","...","return","True","...","...","def","process","(","self",",","row",")",":","...","print","(","row",")","...","...","def","close","(","self",",","error",")",":","...","print","(","``","Closed","with","error",":","%","s","''","%","str","(","error",")",")","...",">",">",">","q","=","df.writeStream.foreach","(","print_row",")",".start","(",")",">",">",">","time.sleep","(","3",")",">",">",">","q.stop","(",")"]
code: def foreach(
      self, f: Union[Callable[[Row], None], "SupportsProcess"]
  ) -> "DataStreamWriter":
      """
      Sets the output of the streaming query to be processed using the provided writer ``f``.
      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.
          This is a simple way to express your processing logic. Note that this does
          not allow you to deduplicate generated data when failures cause reprocessing of
          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.
          The object can have the following methods.
  
          * ``open(partition_id, epoch_id)``: *Optional* method that initializes the processing
              (for example, open a connection, start a transaction, etc). Additionally, you can
              use the `partition_id` and `epoch_id` to deduplicate regenerated data
              (discussed later).
  
          * ``process(row)``: *Non-optional* method that processes each :class:`Row`.
  
          * ``close(error)``: *Optional* method that finalizes and cleans up (for example,
              close connection, commit transaction, etc.) after all rows have been processed.
  
          The object will be used by Spark in the following way.
  
          * A single copy of this object is responsible of all the data generated by a
              single task in a query. In other words, one instance is responsible for
              processing one partition of the data generated in a distributed manner.
  
          * This object must be serializable because each task will get a fresh
              serialized-deserialized copy of the provided object. Hence, it is strongly
              recommended that any initialization for writing data (e.g. opening a
              connection or starting a transaction) is done after the `open(...)`
              method has been called, which signifies that the task is ready to generate data.
  
          * The lifecycle of the methods are as follows.
  
              For each partition with ``partition_id``:
  
              ... For each batch/epoch of streaming data with ``epoch_id``:
  
              ....... Method ``open(partitionId, epochId)`` is called.
  
              ....... If ``open(...)`` returns true, for each row in the partition and
                      batch/epoch, method ``process(row)`` is called.
  
              ....... Method ``close(errorOrNull)`` is called with error (if any) seen while
                      processing rows.
  
          Important points to note:
  
          * The `partitionId` and `epochId` can be used to deduplicate generated data when
              failures cause reprocessing of some input data. This depends on the execution
              mode of the query. If the streaming query is being executed in the micro-batch
              mode, then every partition represented by a unique tuple (partition_id, epoch_id)
              is guaranteed to have the same data. Hence, (partition_id, epoch_id) can be used
              to deduplicate and/or transactionally commit data and achieve exactly-once
              guarantees. However, if the streaming query is being executed in the continuous
              mode, then this guarantee does not hold and therefore should not be used for
              deduplication.
  
          * The ``close()`` method (if exists) will be called if `open()` method exists and
              returns successfully (irrespective of the return value), except if the Python
              crashes in the middle.
  
      .. versionadded:: 2.4.0
  
      .. versionchanged:: 3.5.0
          Supports Spark Connect.
  
      Notes
      -----
      This API is evolving.
  
      Examples
      --------
      >>> import time
      >>> df = spark.readStream.format("rate").load()
  
      Print every row using a function
  
      >>> def print_row(row):
      ...     print(row)
      ...
      >>> q = df.writeStream.foreach(print_row).start()
      >>> time.sleep(3)
      >>> q.stop()
  
      Print every row using a object with process() method
  
      >>> class RowPrinter:
      ...     def open(self, partition_id, epoch_id):
      ...         print("Opened %d, %d" % (partition_id, epoch_id))
      ...         return True
      ...
      ...     def process(self, row):
      ...         print(row)
      ...
      ...     def close(self, error):
      ...         print("Closed with error: %s" % str(error))
      ...
      >>> q = df.writeStream.foreach(print_row).start()
      >>> time.sleep(3)
      >>> q.stop()
      """
  
      from pyspark.rdd import _wrap_function
      from pyspark.serializers import (
          CPickleSerializer,
          AutoBatchedSerializer,
      )
  
      func = self._construct_foreach_function(f)
      serializer = AutoBatchedSerializer(CPickleSerializer())
      wrapped_func = _wrap_function(
          self._spark._sc, func, serializer, serializer
      )
      assert self._spark._sc._jvm is not None
      jForeachWriter = self._spark._sc._jvm.org.apache.spark.sql.execution.python.PythonForeachWriter(
          wrapped_func, self._df._jdf.schema()
      )
      self._jwrite.foreach(jForeachWriter)
      return self
  
imports: import sys
  from collections.abc import Iterator
  from typing import cast, overload, Any, Callable, List, Optional, TYPE_CHECKING, Union
  from py4j.java_gateway import java_import, JavaObject
  from pyspark.sql.column import _to_seq
  from pyspark.sql.readwriter import OptionUtils, to_str
  from pyspark.sql.streaming.query import StreamingQuery
  from pyspark.sql.types import Row, StructType
  from pyspark.sql.utils import ForeachBatchFunction
  from pyspark.errors import PySparkTypeError, PySparkValueError, PySparkAttributeError, PySparkRuntimeError
label_desc: Use image node_id 19 for calling the DataStreamWriter obj's underlying member method code with example usage: obj.foreach(f) and returns: self