azureazure-blob-storageazure-data-factoryazure-data-explorerazureportal
Error while ingesting data from azure append blob to kusto database using an azure data factory
I have an azure append blob(sharing.json) which is of content-type: application/json. I am trying to ingest this into a kusto database with an azure data factory(ADF) but the ingestion is always failing. I get the following error on the output of ADF:
"errors": [
{
"Code": 23302,
"Message": "ErrorCode=KustoWriteFailed,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message=Write to Kusto failed with following error: 'An error occurred for source: 'DataReader'. Error: '''.,Source=Microsoft.DataTransfer.Runtime.KustoConnector,''Type=Kusto.Ingest.Exceptions.IngestClientException,Message=An error occurred for source: 'DataReader'. Error: '',Source=Kusto.Ingest,'",
"EventType": 0,
"Category": 5,
"Data": {},
"MsgId": null,
"ExceptionType": null,
"Source": null,
"StackTrace": null,
"InnerEventInfos": []
}
]
Tried taking help from chatGPT and other online resource but no luck so far.
This is my ADF activity config:
{
"name": "CopyPipeline_k0h",
"properties": {
"activities": [
{
"name": "Copy_k0h",
"type": "Copy",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 3,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [
{
"name": "Source",
"value": "sil-xms-load-max-data//sharing.json"
},
{
"name": "Destination",
"value": "AggregatedSharingTest_v1"
}
],
"typeProperties": {
"source": {
"type": "JsonSource",
"storeSettings": {
"type": "AzureBlobStorageReadSettings",
"recursive": true,
"enablePartitionDiscovery": false
},
"formatSettings": {
"type": "JsonReadSettings"
}
},
"sink": {
"type": "AzureDataExplorerSink",
"ingestionMappingName": "",
"additionalProperties": {
"tags": "drop-by:loadtest",
"format": "multijson"
}
},
"enableStaging": false,
"validateDataConsistency": false,
"logSettings": {
"enableCopyActivityLog": true,
"copyActivityLogSettings": {
"logLevel": "Info",
"enableReliableLogging": true
},
"logLocationSettings": {
"linkedServiceName": {
"referenceName": "LoadTestBlob",
"type": "LinkedServiceReference"
},
"path": "debug-logs"
}
},
"translator": {
"type": "TabularTranslator",
"mappings": [
{
"source": {
"path": "$['deviceId']"
},
"sink": {
"name": "deviceId",
"type": "String"
}
},
{
"source": {
"path": "$['tenant']"
},
"sink": {
"name": "tenant",
"type": "String"
}
},
{
"source": {
"path": "$['tagsSerialNo']"
},
"sink": {
"name": "tagsSerialNo",
"type": "String"
}
},
{
"source": {
"path": "$['metricSum']"
},
"sink": {
"name": "metricSum",
"type": "Int64"
}
},
{
"source": {
"path": "$['metricCount']"
},
"sink": {
"name": "metricCount",
"type": "Int64"
}
},
{
"source": {
"path": "$['notMetricCount']"
},
"sink": {
"name": "notMetricCount",
"type": "Int64"
}
},
{
"source": {
"path": "$['timestamp']"
},
"sink": {
"name": "timestamp",
"type": "DateTime"
}
}
],
"collectionReference": ""
}
},
"inputs": [
{
"referenceName": "SourceDataset_k0h",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "DestinationDataset_k0h",
"type": "DatasetReference"
}
]
}
],
"annotations": [],
"lastPublishTime": "2023-04-18T11:30:35Z"
},
"type": "Microsoft.DataFactory/factories/pipelines"
}
This is the destination dataset config on ADF:
{
"name": "DestinationDataset_k0h",
"properties": {
"linkedServiceName": {
"referenceName": "LoadTestDump",
"type": "LinkedServiceReference"
},
"annotations": [],
"type": "AzureDataExplorerTable",
"schema": [
{
"name": "deviceId",
"type": "string"
},
{
"name": "tenant",
"type": "string"
},
{
"name": "tagsSerialNo",
"type": "string"
},
{
"name": "metricSum",
"type": "long"
},
{
"name": "metricCount",
"type": "long"
},
{
"name": "notMetricCount",
"type": "long"
},
{
"name": "timestamp",
"type": "datetime"
}
],
"typeProperties": {
"table": "AggregatedSharingTest_v1"
}
},
"type": "Microsoft.DataFactory/factories/datasets"
}
This is the Azure blob storage config on ADF:
{
"name": "SourceDataset_k0h",
"properties": {
"linkedServiceName": {
"referenceName": "LoadTestBlob",
"type": "LinkedServiceReference"
},
"annotations": [],
"type": "Json",
"typeProperties": {
"location": {
"type": "AzureBlobStorageLocation",
"fileName": "sharing.json",
"container": "sil-xms-load-max-data"
}
},
"schema": {
"type": "object",
"properties": {
"deviceId": {
"type": "string"
},
"tenant": {
"type": "string"
},
"tagsSerialNo": {
"type": "string"
},
"metricSum": {
"type": "integer"
},
"metricCount": {
"type": "integer"
},
"notMetricCount": {
"type": "integer"
},
"timestamp": {
"type": "string"
}
}
}
},
"type": "Microsoft.DataFactory/factories/datasets"
}
I have tested both the source and destination connections on azure portal and they look good. Not sure what exactly is going wrong since the pipeline runs and run details shows data read and data written but the data is never available on Kusto table for querying and eventually fails with above error
Solution
I tried with your input JSON from storage account and your pipeline JSON and ended up with same error.
In your case, the reason for this error is additionalProperties in the copy activity sink.
When I removed the additionalProperties, I am able to copy the data successfully.
I have 4 rows data in kustos table and you can see two rows inserted from the source using copy activity after removing additonal properties.
Data in target table:
This is my Pipeline JSON for your reference:
{
"name": "pipeline2",
"properties": {
"activities": [
{
"name": "Copy data1",
"type": "Copy",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [
{
"name": "Source",
"value": "data//myjson.json"
},
{
"name": "Destination",
"value": "table1"
}
],
"typeProperties": {
"source": {
"type": "JsonSource",
"storeSettings": {
"type": "AzureBlobFSReadSettings",
"recursive": true,
"enablePartitionDiscovery": false
},
"formatSettings": {
"type": "JsonReadSettings"
}
},
"sink": {
"type": "AzureDataExplorerSink",
"ingestionMappingName": ""
},
"enableStaging": false,
"logSettings": {
"enableCopyActivityLog": true,
"copyActivityLogSettings": {
"logLevel": "Info",
"enableReliableLogging": true
},
"logLocationSettings": {
"linkedServiceName": {
"referenceName": "AzureDataLakeStorage2",
"type": "LinkedServiceReference"
},
"path": "data/debug-logs"
}
},
"translator": {
"type": "TabularTranslator",
"mappings": [
{
"source": {
"path": "$['deviceId']"
},
"sink": {
"name": "deviceId",
"type": "String"
}
},
{
"source": {
"path": "$['tenant']"
},
"sink": {
"name": "tenant",
"type": "Guid"
}
},
{
"source": {
"path": "$['tagsSerialNo']"
},
"sink": {
"name": "tagsSerialNo",
"type": "String"
}
},
{
"source": {
"path": "$['metricSum']"
},
"sink": {
"name": "metricSum",
"type": "Int64"
}
},
{
"source": {
"path": "$['metricCount']"
},
"sink": {
"name": "metricCount",
"type": "Int64"
}
},
{
"source": {
"path": "$['notMetricCount']"
},
"sink": {
"name": "notMetricCount",
"type": "Int64"
}
},
{
"source": {
"path": "$['timestamp']"
},
"sink": {
"name": "timestamp",
"type": "DateTime"
}
}
],
"collectionReference": ""
}
},
"inputs": [
{
"referenceName": "Json1",
"type": "DatasetReference"
}
],
"outputs": [
{
"referenceName": "AzureDataExplorerTable1",
"type": "DatasetReference"
}
]
}
],
"annotations": []
}
}
- Why does my Azure App service have a very slow reponse from an endpoint?
- Creating multiple function apps with one Flex Consumption plan
- ASP.NET Core Authorization with Microsoft Entra ID
- How to get client IP address in Azure Functions C#?
- unable to StartCopyFromUriAsync a blob between 2 storage accounts
- MSgraph cannot find string in attachment value using Azure Automation
- Visual Studio 2022 caches old tenant id, can't get rid of it
- Msgraph-sdk-python get sharepoint site id from name
- using multiple storage accounts locally using azurite at the same time
- Adding Graph API application permission via Terraform shows corrupted IDs instead of scope names
- Azure DevOps Wiki page usage analytics
- How to fix worker runtime metadata for Azure .net8 isolated worker model functions?
- Download attachment content using Microsoft Graph API using Java
- Is it possible to assign a Power BI Pro License to an Azure Service Principal? Deploy .bim in PBI Service without Premium Workspace
- How to grant permissions to my application to use Microsoft Fabric services?
- Azure Devops Pipeline trigger from a different repo does not work
- Issue with adding delegated Graph API permission to Enterprise app with Terraform
- Azure VM metadata missing / emtpy publicIpAddress
- Python: List containers in the Azure Data Lake Storage
- azure blob storage account - log file is not generated as per setup
- SAS token mismatch on Azure DPS with Azure IoT Edge
- Is there a query to run so I can get a list of users who has NOT logged in, in the past 30 days?
- How can I set the MQTT keepalive in the Azure IoTHub C SDK?
- AzureWebJobsStorage Managed Identity not working
- Resource move is not supported for resource types 'microsoft.visualstudio/account'
- Invalid configuration value detected for fs.azure.account.key with com.crealytics:spark-excel
- Access Sharepoint Online Files from .NET Worker Service via Microsoft Graph/OAuth - Unauthorized or Access Denied error (401)
- terraform azure application gateway recreated when add new rule
- Can't add libraries to function_app.py in azure Function
- Azure AI Translation - No document found in source with the given path and filters