> For the complete MongoDB documentation index, see www.mongodb.com/docs/llms.txt

<!--
Tab options on this page. Append to the .md URL to filter:
  ?tabs=<id,...>   select specific tabs (e.g. ?tabs=nodejs,shell)
  ?allTabs=true    include every tab
  (no param)       default: one tab per tabset

Available tabs:
  drivers: java-sync, python, scala
-->

# Write to MongoDB in Streaming Mode

### Java (Sync)

To write data to MongoDB, call the `writeStream()` method on your `Dataset<Row>` object. This method returns a `DataStreamWriter` object, which you can use to specify the format and other configuration settings for your streaming write operation.

You must specify the following configuration settings to write to MongoDB:

| Setting | Description |
| --- | --- |
| `writeStream.format()` | Specifies the format of the underlying output data source. Use `mongodb` to write to MongoDB. |
| `writeStream.option()` | Specifies stream settings, including the MongoDB deployment [connection string](https://www.mongodb.com/docs/manual/reference/connection-string/), MongoDB database and collection, and checkpoint directory. For a list of write stream configuration options, see the [Streaming Write Configuration Options](https://www.mongodb.com/docs/spark-connector/current/streaming-mode/streaming-write-config.md#std-label-spark-streaming-write-conf) guide. |
| `writeStream.outputMode()` | Specifies how data of a streaming DataFrame is written to a streaming sink. To view a list of all supported output modes, see the [Java outputMode documentation.](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/DataStreamWriter.html#outputMode-java.lang.String-) |
| `writeStream.trigger()` | Specifies how often the Spark Connector writes results to the streaming sink. Call this method on the `DataStreamWriter` object you create from the `DataStreamReader` you configure. To use continuous processing, pass `Trigger.Continuous(<time value>)` as an argument, where `<time value>` is how often you want the Spark Connector to asynchronously checkpoint. If you pass any other static method of the `Trigger` class, or if you don't call `writeStream.trigger()`, the Spark connector uses micro-batch processing instead. To view a list of all supported processing policies, see the [Java trigger documentation.](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/Trigger.html) |

The following code snippet shows how to use the previous configuration settings to stream data to MongoDB:

```java
<streaming DataFrame>.writeStream()
  .format("mongodb")
  .option("spark.mongodb.connection.uri", <mongodb-connection-string>)
  .option("spark.mongodb.database", <database-name>)
  .option("spark.mongodb.collection", <collection-name>)
  .outputMode("append");
```

For a complete list of methods, see the [Java Structured Streaming reference.](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/DataStreamWriter.html)

## Example

The following example shows how to stream data from a CSV (comma-separated values) file to MongoDB:

### Java (Sync)

1. Create a `DataStreamReader` object that reads from the CSV file.

2. To create a `DataStreamWriter` object, call the `writeStream()` method on the streaming `Dataset<Row>` that you created with the `DataStreamReader`. Use the `format()` method to specify `mongodb` as the underlying data format.

3. Call the `start()` method on the `DataStreamWriter` object to begin the stream.

As the connector reads data from the CSV file, it adds that data to MongoDB according to the `outputMode` you specify.

```java
// create a local SparkSession
SparkSession spark = SparkSession.builder()
  .appName("writeExample")
  .master("spark://spark-master:<port>")
  .config("spark.jars", "<mongo-spark-connector-JAR-file-name>")
  .getOrCreate();

// define a streaming query
DataStreamWriter<Row> dataStreamWriter = spark.readStream()
  .format("csv")
  .option("header", "true")
  .schema("<csv-schema>")
  .load("<csv-file-name>")
  // manipulate your streaming data
  .writeStream()
  .format("mongodb")
  .option("checkpointLocation", "/tmp/")
  .option("forceDeleteTempCheckpointLocation", "true")
  .option("spark.mongodb.connection.uri", "<mongodb-connection-string>")
  .option("spark.mongodb.database", "<database-name>")
  .option("spark.mongodb.collection", "<collection-name>")
  .outputMode("append");

// run the query
StreamingQuery query = dataStreamWriter.start();
```

## API Documentation

To learn more about the types used in these examples, see the following Apache Spark API documentation:

### Java (Sync)

- [Dataset\<T>](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Dataset.html)

- [DataStreamReader](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/DataStreamReader.html)

- [DataStreamWriter](https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/DataStreamWriter.html)