none
Multiple tables query expression copy?

    Question

  •  

    i have seen the ADF examples of multiple table copy, which internally using Foreach component. 
    In my case, Need to copy the multiple condition based tables through query experission  to corresponding target tables,  instead of creating the multiple copy component. 

    For example, when i input below 2 queries in copy wizard - User query page, it's treating  single custom query with single destination table not 2 tables.

    select empno, enam from emp where modifiedon > '01/01/2018';
    select offno, offname from office where modifiedon > '01/04/2018';

    Wednesday, September 5, 2018 12:51 PM

All replies

  • Hi,

    Yes, in copy wizard, when you input a custom query, it reacts on the single table you selected.

    As you mentioned, in your case you could use a foreach activity in ADFv2, construct an items array storing your custom queries and involves a copy activity in foreach activity to iterate on the items array.

    Hope it'll help. Thanks. 


    • Edited by Wang Zhang Wednesday, September 5, 2018 1:42 PM
    Wednesday, September 5, 2018 1:41 PM
  • i tried with foreach activity but no success on custom query. 

    By any chance, can you send sample json code please for achieving the custom query. 

    Wednesday, September 5, 2018 3:37 PM
  • Sure, here's the pipeline and dataset json definition you could take a reference, in this sample, we copy data in two source tables ( "[dbo].[test-1]" and ["dbo].[test-2]") using two custom queries to two destination tables.

    Pipeline:

    {

        "name": "CopyPipeline_gcf",

        "properties": {

            "activities": [

                {

                    "name": "ForEach_gcf",

                    "type": "ForEach",

                    "typeProperties": {

                        "items": {

                            "value": "@pipeline().parameters.cw_items",

                            "type": "Expression"

                        },

                        "activities": [

                            {

                                "name": "Copy_gcf",

                                "type": "Copy",

                                "policy": {

                                    "timeout": "7.00:00:00",

                                    "retry": 0,

                                    "retryIntervalInSeconds": 30,

                                    "secureOutput": false,

                                    "secureInput": false

                                },

                                "userProperties": [

                                    {

                                        "name": "Source",

                                        "value": "@{item().source.tableName}"

                                    },

                                    {

                                        "name": "Destination",

                                        "value": "@{item().destination.tableName}"

                                    }

                                ],

                                "typeProperties": {

                                    "source": {

                                        "type": "SqlSource",

                                        "sqlReaderQuery": {

                                            "value": "@item().copySource.sqlReaderQuery",

                                            "type": "Expression"

                                        }

                                    },

                                    "sink": {

                                        "type": "SqlSink",

                                        "writeBatchSize": 10000

                                    },

                                    "enableStaging": false,

                                    "dataIntegrationUnits": 0

                                },

                                "inputs": [

                                    {

                                        "referenceName": "SourceDataset_gcf",

                                        "type": "DatasetReference",

                                        "parameters": {

                                            "cw_tableName": "@item().source.tableName"

                                        }

                                    }

                                ],

                                "outputs": [

                                    {

                                        "referenceName": "DestinationDataset_gcf",

                                        "type": "DatasetReference",

                                        "parameters": {

                                            "cw_tableName": "@item().destination.tableName"

                                        }

                                    }

                                ]

                            }

                        ]

                    }

                }

            ],

            "parameters": {

                "cw_items": {

                    "type": "Array",

                    "defaultValue": [

                        {

                            "source": {

                                "tableName": "[dbo].[test-1]"

                            },

                            "destination": {

                                "tableName": "[dbo].[test-1]"

                            },

                            "copySource": {

                                "sqlReaderQuery": "select Prob_2 from [dbo].[test-1] where Prob_1 < '2018-09-06 02:36'"

                            }

                        },

                        {

                            "source": {

                                "tableName": "[dbo].[test-2]"

                            },

                            "destination": {

                                "tableName": "[dbo].[test-2]"

                            },

                            "copySource": {

                                "sqlReaderQuery": "select Prob_2 from [dbo].[test-2] where Prob_1< '2018-09-06 02:36'"

                            }

                        }

                    ]

                }

            }

        },

        "type": "Microsoft.DataFactory/factories/pipelines"

    }

    Source dataset:

    {

        "name": "SourceDataset_gcf",

        "properties": {

            "linkedServiceName": {

                "referenceName": "AzureSql1",

                "type": "LinkedServiceReference"

            },

            "parameters": {

                "cw_tableName": {

                    "type": "String"

                }

            },

            "type": "AzureSqlTable",

            "typeProperties": {

                "tableName": {

                    "value": "@dataset().cw_tableName",

                    "type": "Expression"

                }

            }

        },

        "type": "Microsoft.DataFactory/factories/datasets"

    }

    Sink dataset:

    {
        "name": "DestinationDataset_gcf",
        "properties": {
            "linkedServiceName": {
                "referenceName": "AzureSql1",
                "type": "LinkedServiceReference"
            },
            "parameters": {
                "cw_tableName": {
                    "type": "String"
                }
            },
            "type": "AzureSqlTable",
            "typeProperties": {
                "tableName": {
                    "value": "@dataset().cw_tableName",
                    "type": "Expression"
                }
            }
        },
        "type": "Microsoft.DataFactory/factories/datasets"
    }

    Thursday, September 6, 2018 3:05 AM
  • Thanks Wang. it's very helpful. I've other requirement, instead of sql to sql custom query, now it's dynamics 365 to SQL  custom query and dynamics source reading using fetchxml. Copied below 2 entity fetchxml

    Fetch xml code

    <fetch version="1.0" output-format="xml-platform" mapping="logical" distinct="false">
      <entity name="appointment">
        <attribute name="subject" />
        <attribute name="statecode" />
        <attribute name="modifiedon"  />

    <filter type="and">
     <condition attribute ="modifiedon" operator="gt" value="2018-06-10"/>
    </filter>
              </entity>
    </fetch>


    <fetch version="1.0" output-format="xml-platform" mapping="logical" distinct="false">
      <entity name="account">
        <attribute name="accno" />
        <attribute name="dstatecode" />
        <attribute name="modifiedon"  />

    <filter type="and">
     <condition attribute ="modifiedon" operator="gt" value="2018-06-10"/>
    </filter>

             </entity>

    </fetch>

    Sample copy wizard code from dynamics to SQL but no custom query. In below code how to integrate the above fetchxml's where there is no sqlreaderquery.  Appreciate your advise please

    {
        "name": "CopyPipeline_6hh",
        "properties": {
            "activities": [
                {
                    "name": "ForEach_6hh",
                    "type": "ForEach",
                    "typeProperties": {
                        "items": {
                            "value": "@pipeline().parameters.cw_items",
                            "type": "Expression"
                        },
                        "activities": [
                            {
                                "name": "Copy_6hh",
                                "type": "Copy",
                                "policy": {
                                    "timeout": "7.00:00:00",
                                    "retry": 0,
                                    "retryIntervalInSeconds": 30,
                                    "secureOutput": false,
                                    "secureInput": false
                                },
                                "userProperties": [
                                    {
                                        "name": "Source",
                                        "value": "@{item().source.entityName}"
                                    },
                                    {
                                        "name": "Destination",
                                        "value": "@{item().destination.tableName}"
                                    }
                                ],
                                "typeProperties": {
                                    "source": {
                                        "type": "DynamicsSource"
                                    },
                                    "sink": {
                                        "type": "SqlSink",
                                        "writeBatchSize": 10000
                                    },
                                    "enableStaging": false,
                                    "dataIntegrationUnits": 0
                                },
                                "inputs": [
                                    {
                                        "referenceName": "SourceDataset_6hh",
                                        "type": "DatasetReference",
                                        "parameters": {
                                            "cw_structure": "@item().source.structure",
                                            "cw_entityName": "@item().source.entityName"
                                        }
                                    }
                                ],
                                "outputs": [
                                    {
                                        "referenceName": "DestinationDataset_6hh",
                                        "type": "DatasetReference",
                                        "parameters": {
                                            "cw_structure": "@item().destination.structure",
                                            "cw_tableName": "@item().destination.tableName"
                                        }
                                    }
                                ]
                            }
                        ]
                    }
                }
            ],
            "parameters": {
                "cw_items": {
                    "type": "Array",
                    "defaultValue": [
                        {
                            "source": {
                                "structure": [
                                    {
                                        "name": "address2_addresstypecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "merged",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "accountnumber",
                                        "type": "String"
                                    },
                                    {
                                        "name": "territorycode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_type_is_employer",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "parentaccountid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "name",
                                        "type": "String"
                                    },
                                    {
                                        "name": "opendeals",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "owningbusinessunit",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "owninguser",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "openrevenue_state",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_sales_sendexpiryupgradeemail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_languagepreference",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "accountratingcode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "donotphone",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "preferredcontactmethodcode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "ownerid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "accountclassificationcode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "customersizecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "openrevenue",
                                        "type": "Decimal"
                                    },
                                    {
                                        "name": "donotpostalmail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "openrevenue_date",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "openrevenue_base",
                                        "type": "Decimal"
                                    },
                                    {
                                        "name": "businesstypecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "donotemail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "address2_shippingmethodcode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_sales_sendrenewalemail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_brand",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "address1_addressid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "address2_freighttermscode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "statuscode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "createdon",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "opendeals_state",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_doctorservicegroup",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "donotsendmm",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "donotfax",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "donotbulkpostalmail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_ismigrated",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "modifiedon",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "creditonhold",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "transactioncurrencyid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "accountid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "donotbulkemail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_slug",
                                        "type": "String"
                                    },
                                    {
                                        "name": "exchangerate",
                                        "type": "Decimal"
                                    },
                                    {
                                        "name": "modifiedby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "shippingmethodcode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "createdby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "participatesinworkflow",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "statecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "address2_addressid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "opendeals_date",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_org_capped_unlimited_appointments",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_totalactivated",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "marketingonly",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_defaultaccountplan",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_activatedprimaries",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "cdi_allowtextmessages",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_migrationkey",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_totalleads",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_registeredpercentage",
                                        "type": "Decimal"
                                    },
                                    {
                                        "name": "bu_org_dependant_paid_for_already",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_org_invoicing_approach",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "cdi_throttlesend",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "followemail",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_activateddependants",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_org_end_of_contract_renewal_date_m",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "description",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_org_approx_number_of_core_lives",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_org_date_data_expected",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "timezoneruleversionnumber",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_contractstartdate",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_org_actual_date_and_time_first_ema",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_org_date_and_time_sending_of_email",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "importsequencenumber",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_verificationstatus",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_industry",
                                        "type": "Guid"
                                    }
                                ],
                                "entityName": "account"
                            },
                            "destination": {
                                "structure": [
                                    {
                                        "name": "openrevenue_date",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "territorycode",
                                        "type": "Int32"
                                    }
                                ],
                                "tableName": "[dbo].[account]"
                            },
                            "copyActivity": {
                                "translator": {
                                    "type": "TabularTranslator",
                                    "columnMappings": {
                                        "openrevenue_date": "openrevenue_date",
                                        "territorycode": "territorycode"
                                    }
                                }
                            }
                        },
                        {
                            "source": {
                                "structure": [
                                    {
                                        "name": "actualend",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "prioritycode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_acceptedukjurisdiction",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_bookedby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "isworkflowcreated",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "activityid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_offeredprivategpreferral",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_doctorgenderselection",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_triagedemergencycallsuccess",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "attachmentcount",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_referencenumber",
                                        "type": "String"
                                    },
                                    {
                                        "name": "owningbusinessunit",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_doctor",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "owninguser",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "regardingobjectid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "ownerid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_ismigrated",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_patientaddress",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_bookingstatus",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_patient",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "attachmenterrors",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "scheduledend",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_patientattendedappointment",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_requestedphonenumberprefix",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_originalstarttime",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_testimonialconsent",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "createdon",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_followupbooked",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "timezoneruleversionnumber",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "ismapiprivate",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "statuscode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_webrequestedstart",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_patientpostcode",
                                        "type": "String"
                                    },
                                    {
                                        "name": "isregularactivity",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "isbilled",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "scheduledstart",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_bookedon",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_appointmenttype",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "instancetypecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_policyid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_triagedsmssent",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_cancellationreason",
                                        "type": "String"
                                    },
                                    {
                                        "name": "subject",
                                        "type": "String"
                                    },
                                    {
                                        "name": "isalldayevent",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_symptomdescription",
                                        "type": "String"
                                    },
                                    {
                                        "name": "modifiedby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_requestedtimezone",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "createdby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_doctorexperiencedconsultationproblems",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_healthconcern",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_isdemo",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "modifiedonbehalfby",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_originaldoctorid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_otherhealthconcern",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_declinedprivategpreferral",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "statecode",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "scheduleddurationminutes",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "activitytypecode",
                                        "type": "String"
                                    },
                                    {
                                        "name": "modifiedon",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_requestedphonenumber",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_feedbackprovided",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_sharenoteswithgp",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_addtionalnotes",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_sessionnotes",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_familyhistory",
                                        "type": "String"
                                    },
                                    {
                                        "name": "actualstart",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_history",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_differentialdiagnosis",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_sharenoteswithguardians",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_presentingcomplainthistory",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_introduction",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_recommendedactions",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_feedbackskipped",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_allergies",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_medications",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_triagedemergencycalltime",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_triagingdoctorid",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_triagedtime",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_completeddate",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_signoff",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_feedback",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_survey_ease",
                                        "type": "Double"
                                    },
                                    {
                                        "name": "bu_survey_advice",
                                        "type": "Double"
                                    },
                                    {
                                        "name": "bu_survey_satisfaction",
                                        "type": "Double"
                                    },
                                    {
                                        "name": "bu_survey_recommend",
                                        "type": "Double"
                                    },
                                    {
                                        "name": "bu_survey_technology",
                                        "type": "Double"
                                    },
                                    {
                                        "name": "bu_otherpresentingcomplaint",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_speakerstestpassed",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_presentingcomplaint",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_webcamtestpassed",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_connectiontestpassed",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "bu_microphonetestpassed",
                                        "type": "Boolean"
                                    },
                                    {
                                        "name": "owningteam",
                                        "type": "Guid"
                                    },
                                    {
                                        "name": "bu_failuretoattendnotes",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_followuppathway",
                                        "type": "Int32"
                                    },
                                    {
                                        "name": "bu_publisheddate",
                                        "type": "DateTime"
                                    },
                                    {
                                        "name": "bu_followupdue",
                                        "type": "DateTime"
                                    }
                                ],
                                "entityName": "appointment"
                            },
                            "destination": {
                                "structure": [
                                    {
                                        "name": "bu_requestedphonenumber",
                                        "type": "String"
                                    },
                                    {
                                        "name": "bu_bookedby",
                                        "type": "Guid"
                                    }
                                ],
                                "tableName": "[dbo].[appointment]"
                            },
                            "copyActivity": {
                                "translator": {
                                    "type": "TabularTranslator",
                                    "columnMappings": {
                                        "bu_requestedphonenumber": "bu_requestedphonenumber",
                                        "bu_bookedby": "bu_bookedby"
                                    }
                                }
                            }
                        }
                    ]
                }
            }
        },
        "type": "Microsoft.DataFactory/factories/pipelines"
    }
    
    
    
    
    source 
    
    {
        "name": "SourceDataset_6hh",
        "properties": {
            "linkedServiceName": {
                "referenceName": "Dynamics_DCA_UAT",
                "type": "LinkedServiceReference"
            },
            "parameters": {
                "cw_structure": {
                    "type": "Array"
                },
                "cw_entityName": {
                    "type": "String"
                }
            },
            "type": "DynamicsEntity",
            "structure": {
                "value": "@dataset().cw_structure",
                "type": "Expression"
            },
            "typeProperties": {
                "entityName": {
                    "value": "@dataset().cw_entityName",
                    "type": "Expression"
                }
            }
        },
        "type": "Microsoft.DataFactory/factories/datasets"
    }
    
    
    
    destination
    
    
    
    {
        "name": "DestinationDataset_6hh",
        "properties": {
            "linkedServiceName": {
                "referenceName": "AzureSqlDatabase51",
                "type": "LinkedServiceReference"
            },
            "parameters": {
                "cw_structure": {
                    "type": "Array"
                },
                "cw_tableName": {
                    "type": "String"
                }
            },
            "type": "AzureSqlTable",
            "structure": {
                "value": "@dataset().cw_structure",
                "type": "Expression"
            },
            "typeProperties": {
                "tableName": {
                    "value": "@dataset().cw_tableName",
                    "type": "Expression"
                }
            }
        },
        "type": "Microsoft.DataFactory/factories/datasets"
    }
    
    
    
    
    
    
    

    Thursday, September 6, 2018 9:32 PM
  • Hi,

    To integrate custom query in pipeline, you need to modify pipeline definition by adding "copySource" property into array "cw_items", give it a default value (your custom query listed above) and then refer the "copySource" in the copy activity settings, this is exactly the same as the sql to sql copy using custom query we discussed before, please refer the following pipeline json definition(your dataset definition is right and need no change):

    {
        "name": "CopyPipeline_zy3",
        "properties": {
            "activities": [
                {
                    "name": "ForEach_zy3",
                    "type": "ForEach",
                    "typeProperties": {
                        "items": {
                            "value": "@pipeline().parameters.cw_items",
                            "type": "Expression"
                        },
                        "activities": [
                            {
                                "name": "Copy_zy3",
                                "type": "Copy",
                                "userProperties": [
                                    {
                                        "name": "Source",
                                        "value": "@{item().source.entityName}"
                                    },
                                    {
                                        "name": "Destination",
                                        "value": "@{item().destination.tableName}"
                                    }
                                ],
                                "typeProperties": {
                                    "source": {
                                        "type": "DynamicsSource",
                                        "query": {
                                            "value": "@item().copySource.query",
                                            "type": "Expression"
                                        }
                                    },
                                    "sink": {
                                        "type": "SqlSink",
                                        "writeBatchSize": 10000
                                    },
                                    "enableStaging": false,
                                    "dataIntegrationUnits": 0
                                },
                                "inputs": [
                                    {
                                        "referenceName": "SourceDataset_zy3",
                                        "type": "DatasetReference",
                                        "parameters": {
                                            "cw_structure": "@item().source.structure",
                                            "cw_entityName": "@item().source.entityName"
                                        }
                                    }
                                ],
                                "outputs": [
                                    {
                                        "referenceName": "DestinationDataset_zy3",
                                        "type": "DatasetReference",
                                        "parameters": {
                                            "cw_structure": "@item().destination.structure",
                                            "cw_tableName": "@item().destination.tableName"
                                        }
                                    }
                                ]
                            }
                        ]
                    }
                }
            ],
            "parameters": {
                "cw_items": {
                    "type": "Array",
                    "defaultValue": [
                        {
                            "source": {
                                "structure": [
                                    // your first source structure
                                ],
                                "entityName": // your first source entityName
                            },
                            "destination": {
                                "structure": [
                                    // your first sink structure
                                ],
                                "tableName": // your first sink tableName
                            },
                            "copySource": {
                                "query": // your first source Fetch xml query
                            },
                            "copyActivity": {
                                "translator": {
                                    // your first copy activity column mapping
                                }
                            }
                        },
                        {
                            "source": {
                                "structure": [
                                    // your second source structure
                                ],
                                "entityName": // your second source entityName
                            },
                            "destination": {
                                "structure": [
                                    // your second sink structure
                                ],
                                "tableName": // your second sink tableName
                            },
                            "copySource": {
                                "query": // your second source Fetch xml query
                            },
                            "copyActivity": {
                                "translator": {
                                    // your second copy activity column mapping
                                }
                            }
                        }
                    ]
                }
            }
        },
        "type": "Microsoft.DataFactory/factories/pipelines"
    }

    Thanks.
    Friday, September 7, 2018 3:07 AM