cosmosDB Azure DataFactor DataFlow configure sink without a throughput value

Question

I have a simple dataflow in a pipeline that reads a tab delimited file with 5 columns. The last column is valid json. I want the data flow to read the file and insert each json in each row in column 5 to cosmosDB. All of the connections pass test connection and I can preview the data in the data sink. However, when I run the dataflow in the pipeline I get the following error: It appears

to because I am not setting a value for throughput. However this is a serverlaess instance of cosmosdb and if you set a value for threshold then it errors out saying you can't set a value for threshold.

Anyone know how to set a sink up for a serverless cosmosDB NoSQL instance ?

Thanks

Operation on target LoadOpenLibraryDumps failed: {"StatusCode":"DFExecutorUserError","Message":"Job failed due to reason: at Sink 'sink1': 1","Details":"java.lang.ArrayIndexOutOfBoundsException: 1\n\tat com.microsoft.dataflow.store.cosmosdb.CosmosDBStore$.setThroughput(CosmosDBStore.scala:914)\n\tat com.microsoft.dataflow.store.cosmosdb.CosmosDBWriter.$anonfun$write$12(CosmosDBStore.scala:352)\n\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n\tat scala.util.Try$.apply(Try.scala:213)\n\tat com.microsoft.dataflow.store.cosmosdb.CosmosDBWriter.$anonfun$write$11(CosmosDBStore.scala:267)\n\tat com.microsoft.dataflow.FlowRunner$$anon$2.$anonfun$visitTransforms$14(FlowRunner.scala:1214)\n\tat com.microsoft.dataflow.FlowCode.$anonfun$run$13(FlowRunner.scala:240)\n\tat scala.concurrent.Future$.$anonfun$apply$1(Future.scala:659)\n\tat scala.util.Success.$anonfun$map$1(Try.scala:255)\n\tat scala.util.Success.map(Try.scala:213)\n\tat scala.concurrent.Future.$anonfun$map$1(Future.scala:292)\n\tat scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33)\n\tat scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33)\n\tat scala.concurrent.impl.CallbackRunn"}

Json for DataFlow Definition

{
    "name": "LoadOpenLibraryDumps",
    "properties": {
        "description": "loads dump files from openlibrary editions, authors, works, etc...",
        "type": "MappingDataFlow",
        "typeProperties": {
            "sources": [
                {
                    "dataset": {
                        "referenceName": "SourceDataset_sxx",
                        "type": "DatasetReference"
                    },
                    "name": "BlobForOpenLibrary"
                }
            ],
            "sinks": [
                {
                    "dataset": {
                        "referenceName": "CosmosDbNoSqlContainer3",
                        "type": "DatasetReference"
                    },
                    "name": "sink1"
                }
            ],
            "transformations": [
                {
                    "name": "getContainer"
                },
                {
                    "name": "idValue"
                },
                {
                    "name": "infoToInsertToCosmosDB"
                },
                {
                    "name": "jsonToInsert"
                },
                {
                    "name": "select2"
                },
                {
                    "name": "partitionKeyValue"
                },
                {
                    "name": "finalJSon"
                }
            ],
            "scriptLines": [
                "parameters{",
                "     sinkContainerName as string (container)",
                "}",
                "source(output(",
                "          Column_1 as string,",
                "          Column_2 as string,",
                "          Column_3 as string,",
                "          Column_4 as string,",
                "          Column_5 as string",
                "     ),",
                "     allowSchemaDrift: true,",
                "     validateSchema: false,",
                "     ignoreNoFilesFound: false) ~> BlobForOpenLibrary",
                "BlobForOpenLibrary derive(container = replace (Column_1, '/type/','')) ~> getContainer",
                "getContainer derive(idValue = regexReplace(\r",
                "  Column_2, \r",
                "  '/.*/','')) ~> idValue",
                "jsonToInsert select(mapColumn(",
                "          container,",
                "          jsonToInsert = jsonToOutput,",
                "          partionKey = partionKeyValue",
                "     ),",
                "     skipDuplicateMapInputs: true,",
                "     skipDuplicateMapOutputs: true) ~> infoToInsertToCosmosDB",
                "select2 derive(jsonToOutput = concat(substring(Column_5,1,length(Column_5)-1),',\"id\":\"',idValue,'\"}')) ~> jsonToInsert",
                "partitionKeyValue select(mapColumn(",
                "          Column_5,",
                "          container,",
                "          idValue,",
                "          partionKeyValue",
                "     ),",
                "     skipDuplicateMapInputs: true,",
                "     skipDuplicateMapOutputs: true) ~> select2",
                "idValue derive(partionKeyValue = iif (instr(Column_5,'publish_date')>0 , regexExtract(Column_5, '\"publish_date\"\\\\s*:\\\\s*\"(\\\\d{4})\"', 1),'/id')) ~> partitionKeyValue",
                "infoToInsertToCosmosDB select(mapColumn(",
                "          jsonToInsert",
                "     ),",
                "     skipDuplicateMapInputs: true,",
                "     skipDuplicateMapOutputs: true) ~> finalJSon",
                "finalJSon sink(allowSchemaDrift: true,",
                "     validateSchema: false,",
                "     deletable:false,",
                "     insertable:true,",
                "     updateable:false,",
                "     upsertable:false,",
                "     format: 'document',",
                "     skipDuplicateMapInputs: true,",
                "     skipDuplicateMapOutputs: true) ~> sink1"
            ]
        }
    }
}

Json to be inserted

{
  "type": { "key": "/type/author" },
  "name": "Kathy Godhy",
  "key": "/authors/OL10000125A",
  "source_records": [ "bwb:9786155232114" ],
  "latest_revision": 1,
  "revision": 1,
  "created": {
    "type": "/type/datetime",
    "value": "2021-12-26T21:31:59.105206"
  },
  "last_modified": {
    "type": "/type/datetime",
    "value": "2021-12-26T21:31:59.105206"
  },
  "id": "OL10000125A"
}

Answer 1

Hello @Brady Chambers,

As per this MS Documentation,

Currently, The Azure Cosmos DB serverless is not supported in mapping data flow.

As you are using the Serverless Cosmos DB, this might be the reason for the above behavior.

You can try the below possible workarounds to achieve your requirement.

When looking at your last column output data, you are getting the JSON string as output. So, there are two Scenarios that you can consider here.

1. Whether you need the JSON string data to be copied as a single string column of the new document in your cosmos sink data.
2. Or it needed to be copied as a proper JSON document.

Solution 1 for the First Scenario:

As the serverless cosmos db is not supported in mapping dataflow, take a temporary Delimited file dataset with any special character like | or tab (special characters which does not exist in your JSON sink) as a column delimiter.

Give this dataset as dataflow sink. In the dataflow sink, go to settings -> output to single file -> give file name. This generates only a single file, but you can avoid it if not needed for your requirement.

Now, in the pipeline, take a Copy activity after the dataflow activity. In the copy activity, take the above delimited text dataset as source and your serverless cosmos db dataset as sink. Select the required Write behavior.

Upon the running the pipeline, first the JSON strings data will be copied to the temporary delimited file and then the copy activity will copy the same to the required cosmos target.

Solution 2 for the Second Scenario:

Here, after the above steps in your mapping dataflow, take a Parse transformation and parse the JSON string in each row to a proper JSON object. You need to define the schema here. After that, take a temporary JSON file dataset as dataflow sink and follow the same above steps if you need it as a single file.

After the dataflow activity, take Copy activity and give the above JSON dataset as source and your cosmos db dataset as target. Set the required mapping in the copy activity mapping section.

This will generate the required JSON document in the target cosmos database.

Hope this helps.

If the answer is helpful, please click Accept Answer and kindly upvote it. If you have any further questions about this answer, please click Comment