在Azure数据工厂管道中，我试图将参数传递给Azure SQL中的存储过程（下面的步骤2）。但我找不到怎么样。有些人知道如何将参数列表（从输入数据集）传递到存储过程（outputdataset）并让存储过程完成工作吗？（在这种情况下，upsert）
规格
1.Input数据集从blob存储中获取json（它的列表）。 2. Pipeline使用步骤1中的参数调用存储过程。
管道
{
"name": "SprocActivitySamplePipeline",
"properties": {
"activities": [
{
"type": "SqlServerStoredProcedure",
"typeProperties": {
"storedProcedureName": "UpsesrtSensorRecords",
"storedProcedureParameters": {
"device_id": "*device_id from input dataset",
"user_id": "1",
"serial": "serial from input dataset",
"measured_time": "$$Text.Format('{0:yyyy-MM-dd HH:mm:ss}', SliceStart)"
}
},
"sink" : {
"type": "SqlSink"
},
"inputs": [
{
"name": "SimpleCopyInputDataset"
}
],
"outputs": [
{
"name": "AzureSQLSprocOut"
}
],
"scheduler": {
"frequency": "Day",
"interval": 1
},
"name": "SprocActivitySample"
}
],
"start": "2017-04-04T00:00:00Z",
"end": "2017-04-04T05:00:00Z",
"isPaused": false,
"hubName": "yoshinobufuruyadf_hub",
"pipelineMode": "Scheduled"
}
}
输入数据集
{
"name": "SimpleCopyInputDataset",
"properties": {
"structure": [
{
"name": "device_id",
"type": "String"
},
{
"name": "user_id",
"type": "Int64"
},
{
"name": "serial",
"type": "String"
},
{
"name": "longitude",
"type": "Double"
},
{
"name": "latitude",
"type": "Double"
},
{
"name": "location_name",
"type": "String"
},
{
"name": "measured_time",
"type": "String"
},
{
"name": "water_level",
"type": "Double"
},
{
"name": "water_temperature",
"type": "Double"
},
{
"name": "temperature",
"type": "Double"
},
{
"name": "relative_temperature",
"type": "Double"
},
{
"name": "soil_temperature",
"type": "Double"
},
{
"name": "soil_moisture",
"type": "Double"
},
{
"name": "soil_ec",
"type": "Double"
},
{
"name": "water_ec",
"type": "Double"
},
{
"name": "leaf_wetting",
"type": "Double"
},
{
"name": "soil_temperature_5cm",
"type": "Double"
},
{
"name": "soil_temperature_10cm",
"type": "Double"
},
{
"name": "external_solar_radiation",
"type": "Double"
},
{
"name": "salt_concentration",
"type": "Double"
},
{
"name": "weather_wind_direction",
"type": "Double"
},
{
"name": "weather_wind_speed",
"type": "Double"
},
{
"name": "weather_instant_wind_speed",
"type": "Double"
},
{
"name": "weather_rainfall",
"type": "Double"
},
{
"name": "weather_temperature",
"type": "Double"
},
{
"name": "weather_relative_temperature",
"type": "Double"
},
{
"name": "weather_ultraviolet_ray_amount",
"type": "Double"
},
{
"name": "weather_illuminance",
"type": "Int64"
}
],
"published": false,
"type": "AzureBlob",
"linkedServiceName": "AzureStorageLinkedService",
"typeProperties": {
"fileName": "NormalizedData.json",
"folderPath": "imported-json",
"format": {
"type": "JsonFormat",
"jsonNodeReference": "$.['sensors']",
"jsonPathDefinition": {
"device_id": "['device_id']",
"user_id": "['user_id']",
"serial": "['serial']",
"longitude": "['longitude']",
"latitude": "['latitude']",
"location_name": "['location_name']",
"measured_time": "['measured_time']",
"water_level": "['water_level']",
"water_temperature": "['water_temperature']",
"temperature": "['temperature']",
"relative_temperature": "['relative_temperature']",
"soil_temperature": "['soil_temperature']",
"soil_moisture": "['soil_moisture']",
"soil_ec": "['soil_ec']",
"water_ec": "['water_ec']",
"leaf_wetting": "['leaf_wetting']",
"soil_temperature_5cm": "['soil_temperature_5cm']",
"soil_temperature_10cm": "['soil_temperature_10cm']",
"external_solar_radiation": "['external_solar_radiation']",
"salt_concentration": "['salt_concentration']",
"weather_wind_direction": "['weather_wind_direction']",
"weather_wind_speed": "['weather_wind_speed']",
"weather_instant_wind_speed": "['weather_instant_wind_speed']",
"weather_rainfall": "['weather_rainfall']",
"weather_temperature": "['weather_temperature']",
"weather_relative_temperature": "['weather_relative_temperature']",
"weather_ultraviolet_ray_amount": "['weather_ultraviolet_ray_amount']",
"weather_illuminance": "['weather_illuminance']"
}
}
},
"availability": {
"frequency": "Day",
"interval": 1
},
"external": true,
"policy": {}
}
}
存储过程
Create PROCEDURE UpsesrtSensorRecords
(
@device_id INT,
@user_id INT,
@serial VARCHAR(16),
@longitude FLOAT = NULL,
@latitude FLOAT = NULL,
@location_name VARCHAR(128) = NULL,
@measured_time DATETIME,
@water_level FLOAT = NULL,
@water_temperature FLOAT = NULL,
@temperature FLOAT = NULL,
@relative_temperature FLOAT = NULL,
@soil_temperature FLOAT = NULL,
@soil_moisture FLOAT = NULL,
@soil_ec FLOAT = NULL,
@water_ec FLOAT = NULL,
@leaf_wetting FLOAT = NULL,
@soil_temperature_5cm FLOAT = NULL,
@soil_temperature_10cm FLOAT = NULL,
@external_solar_radiation FLOAT = NULL,
@salt_concentration FLOAT = NULL,
@weather_wind_direction FLOAT = NULL,
@weather_wind_speed FLOAT = NULL,
@weather_instant_wind_speed FLOAT = NULL,
@weather_rainfall FLOAT = NULL,
@weather_temperature FLOAT = NULL,
@weather_relative_temperature FLOAT = NULL,
@weather_ultraviolet_ray_amount FLOAT = NULL,
@weather_illuminance INT = NULL
)
AS
DECLARE @RowCount INT;
BEGIN
SET @RowCount = (SELECT count(*) FROM dbo.T_SensorRecords WHERE (device_id = @device_id)AND (measured_time = @measured_time));
IF @RowCount=0
INSERT INTO dbo.T_SensorRecords (device_id,user_id,serial,longitude,latitude,location_name,measured_time,water_level ,water_temperature ,temperature ,relative_temperature ,soil_temperature ,soil_moisture ,soil_ec ,water_ec ,leaf_wetting ,soil_temperature_5cm ,soil_temperature_10cm ,external_solar_radiation ,salt_concentration ,weather_wind_direction ,weather_wind_speed ,weather_instant_wind_speed ,weather_rainfall,weather_temperature ,weather_relative_temperature,weather_ultraviolet_ray_amount,weather_illuminance )
VALUES(@device_id,@user_id,@serial,@longitude,@latitude,@location_name,@measured_time,@water_level ,@water_temperature ,@temperature ,@relative_temperature ,@soil_temperature ,@soil_moisture ,@soil_ec ,@water_ec ,@leaf_wetting ,@soil_temperature_5cm ,@soil_temperature_10cm ,@external_solar_radiation ,@salt_concentration ,@weather_wind_direction ,@weather_wind_speed ,@weather_instant_wind_speed ,@weather_rainfall,@weather_temperature ,@weather_relative_temperature,@weather_ultraviolet_ray_amount,@weather_illuminance);
ELSE
UPDATE dbo.T_SensorRecordss
SET user_id=@user_id,serial=@serial,longitude=@longitude,latitude=@latitude,location_name=@location_name,water_level=@water_level,water_temperature=@water_temperature,temperature=@temperature,relative_temperature=@relative_temperature,soil_temperature=@soil_temperature,soil_moisture=@soil_moisture,soil_ec=@soil_ec,water_ec=@water_ec,leaf_wetting=@leaf_wetting,soil_temperature_5cm=@soil_temperature_5cm,soil_temperature_10cm=@soil_temperature_10cm,external_solar_radiation=@external_solar_radiation,salt_concentration=@salt_concentration,weather_wind_direction=@weather_wind_direction,weather_wind_speed=@weather_wind_speed,weather_instant_wind_speed=@weather_instant_wind_speed,weather_rainfall=@weather_rainfall,weather_temperature=@weather_temperature,weather_relative_temperature=@weather_relative_temperature,weather_ultraviolet_ray_amount=@weather_ultraviolet_ray_amount,weather_illuminance=@weather_illuminance
WHERE device_id = @device_id AND measured_time = @measured_time;
END
我发现之前的帖子没有答案非常相似。 Azure data factory SqlSink, use json as input。他似乎没有得到答案。
您需要使用元数据和查找活动对ADFv2执行此操作。然后让存储的proc调用活动依赖于上游结果。
我不能真正评论，因为我没有真正做过，但知道它可能。
希望这可以帮助。