使用Azure Data Factory从Append Blob向Kusto导入数据失败求助

问题详情 #

我有一个内容类型为application/json的Azure Append Blob（sharing.json），尝试通过Azure Data Factory（ADF）将其导入Kusto数据库，但导入始终失败。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": []
        }
    ]

已确认源和目标连接在Azure门户测试正常，管道运行时显示已读取和写入数据，但数据从未出现在Kusto表中，最终抛出上述错误。尝试过ChatGPT及其他在线资源，未能解决问题。

ADF活动配置 #

{
    "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"
}

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"
}

ADF Azure Blob存储配置 #

{
    "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"
}

排查与解决方案 #

1. 验证Blob的JSON格式是否匹配MultiJSON要求 #

Sink中指定了format: multijson，要求Blob内每行是独立的JSON对象，而非包含数组的JSON文档。如果sharing.json是数组格式（如[{"deviceId":"xxx"},{"deviceId":"yyy"}]），ADF无法正确解析，会触发DataReader错误。

• 检查Blob内容，若为数组格式，需修改为每行一个JSON对象；或在JsonReadSettings中添加jsonPathDefinition": "$[*]"，同时将Translator的collectionReference设为$。

2. 修复Timestamp字段类型转换问题 #

源数据中timestamp是字符串类型，Sink映射为DateTime，但Kusto仅支持ISO 8601格式的DateTime（如yyyy-MM-ddTHH:mm:ss.fffZ）。格式不匹配会导致 ingestion 失败。

• 检查Blob中timestamp的格式，确保符合标准；或在ADF中添加转换步骤，将字符串转换为标准DateTime格式后再导入。

3. 查看Kusto侧Ingestion日志获取详细错误 #

当前ADF错误信息模糊，可通过Kusto查询获取具体失败原因：

.show ingestion failures
| where Database == "你的数据库名" and Table == "AggregatedSharingTest_v1"
| order by FailedOn desc

4. 调整ADF复制活动配置 #

• 移除Sink的additionalProperties中的format设置：ADF会自动推断源格式，手动指定可能与实际格式冲突；
• 启用Staging功能：将数据先写入临时Blob，再由Kusto Ingestion Service拉取，提升稳定性且便于排查数据问题；
• 修正Source的JsonReadSettings：若Blob为数组格式，添加如下配置：

"jsonPathDefinition": "$[*]",
"collectionReference": "$"

同时确保Translator的collectionReference设为$。

5. 检查Kusto表权限 #

确认ADF使用的服务主体对目标表有ingestor权限，可通过以下命令验证：

.show table AggregatedSharingTest_v1 principals

若服务主体不在列表中，执行以下命令添加权限：

.add table AggregatedSharingTest_v1 ingestors ('aadapp=<服务主体ID>;<租户ID>')

内容的提问来源于stack exchange，提问作者Dipanshu