streaming-read-config.txt

.. _spark-streaming-read-conf:

====================================
Streaming Read Configuration Options
====================================

.. contents:: On this page
   :local:
   :backlinks: none
   :depth: 1
   :class: singlecol

.. facet::
   :name: genre
   :values: reference
 
.. meta::
   :keywords: change stream, customize

.. _spark-streaming-input-conf:

Overview
--------

You can configure the following properties when reading data from MongoDB in streaming mode.

.. include:: /includes/conf-read-prefix.rst

.. list-table::
   :header-rows: 1
   :widths: 35 65

   * - Property name
     - Description
   
   * - ``connection.uri``
     - | **Required.**
       | The connection string configuration key.
       |
       | **Default:** ``mongodb://localhost:27017/``

   * - ``database``
     - | **Required.**
       | The database name configuration.

   * - ``collection``
     - | **Required.**
       | The collection name configuration.
       | You can specify multiple collections by separating each collection name
         with a comma (,). Do not add a space between the collections
         unless the space is a part of the collection name.
       | Specify multiple collections as follows:

       .. code-block:: java

          ...
          .option("spark.mongodb.collection", "collectionOne,collectionTwo")

       | You can scan from all collections in the database by using an asterisk (*).
       | Specify all collections as follows:

       .. code-block:: java

          ...
          .option("spark.mongodb.collection", "*")

       If you create a collection while scanning from all collections, it is
       automatically picked up for scanning.

   * - ``comment``
     - | The comment to append to the read operation. Comments appear in the 
         :manual:`output of the Database Profiler. </reference/database-profiler>`
       |
       | **Default:** None 

   * - ``mongoClientFactory``
     - | MongoClientFactory configuration key.
       | You can specify a custom implementation, which must implement the
         ``com.mongodb.spark.sql.connector.connection.MongoClientFactory``
         interface.
       |
       | **Default:** ``com.mongodb.spark.sql.connector.connection.DefaultMongoClientFactory``

   * - ``aggregation.pipeline``
     - | Specifies a custom aggregation pipeline to apply to the collection
         before sending data to Spark.
       | The value must be either an extended JSON single document or list
         of documents.
       | A single document resembles the following:

       .. code-block:: json

          {"$match": {"closed": false}}

       | A list of documents resembles the following:

       .. code-block:: json

          [{"$match": {"closed": false}}, {"$project": {"status": 1, "name": 1, "description": 1}}]

       .. important::

          Custom aggregation pipelines must be compatible with the
          partitioner strategy. For example, aggregation stages such as
          ``$group`` do not work with any partitioner that creates more than
          one partition.

   * - ``aggregation.allowDiskUse``
     - | Specifies whether to allow storage to disk when running the
         aggregation.
       |
       | **Default:** ``true``

   * - ``change.stream.``
     - | Change stream configuration prefix.
       | See the
         :ref:`Change Stream Configuration <change-stream-conf>` section for more
         information about change streams.

   * - ``outputExtendedJson``
     - | When ``true``, the connector converts BSON types not supported by Spark into 
         extended JSON strings.
         When ``false``, the connector uses the original relaxed JSON format for 
         unsupported types.
       |
       | **Default:** ``false``

.. _change-stream-conf:

Change Stream Configuration
---------------------------

You can configure the following properties when reading a change stream from MongoDB:

.. list-table::
   :header-rows: 1
   :widths: 35 65

   * - Property name
     - Description

   * - ``change.stream.lookup.full.document``

     - Determines what values your change stream returns on update
       operations.

       The default setting returns the differences between the original
       document and the updated document.

       The ``updateLookup`` setting also returns the differences between the
       original document and updated document, but it also includes a copy of the
       entire updated document.

       **Default:** "default"

       .. tip::

          For more information on how this change stream option works,
          see the MongoDB server manual guide
          :manual:`Lookup Full Document for Update Operation </changeStreams/#lookup-full-document-for-update-operations>`.

   * - ``change.stream.micro.batch.max.partition.count``
     - | The maximum number of partitions the {+connector-short+} divides each 
         micro-batch into. Spark workers can process these partitions in parallel.
       |  
       | This setting applies only when using micro-batch streams.
       |
       | **Default**: ``1``

       .. warning:: Event Order

          Specifying a value larger than ``1`` can alter the order in which
          the {+connector-short+} processes change events. Avoid this setting
          if out-of-order processing could create data inconsistencies downstream. 

   * - ``change.stream.publish.full.document.only``
     - | Specifies whether to publish the changed document or the full
         change stream document.
       |
       | When this setting is ``false``, you must specify a schema. The schema
         must include all fields that you want to read from the change stream. You can
         use optional fields to ensure that the schema is valid for all change-stream
         events.
       |
       | When this setting is ``true``, the connector exhibits the following behavior:
       
       - The connector filters out messages that
         omit the ``fullDocument`` field and publishes only the value of the
         field.
       - If you don't specify a schema, the connector infers the schema
         from the change stream document.

       **Default**: ``false``
       
       .. note::

          This setting overrides the ``change.stream.lookup.full.document``
          setting.

   * - ``change.stream.startup.mode``
     - | Specifies how the connector starts up when no offset is available.
       
       | This setting accepts the following values:
        
       - ``latest``: The connector begins processing
         change events starting with the most recent event.
         It will not process any earlier unprocessed events.
       - ``timestamp``: The connector begins processing change events at a specified time.
           
         To use the ``timestamp`` option, you must specify a time by using the
         ``change.stream.startup.mode.timestamp.start.at.operation.time`` setting.
         This setting accepts timestamps in the following formats:
         
         - An integer representing the number of seconds since the
           :wikipedia:`Unix epoch <Unix_time>`
         - A date and time in
           `ISO-8601 <https://www.iso.org/iso-8601-date-and-time-format.html>`__
           format with one-second precision
         - An extended JSON ``BsonTimestamp``
       
         **Default**: ``latest``
 
Specifying Properties in ``connection.uri``
-------------------------------------------

.. include:: /includes/connection-read-config.rst