I'm trying to do an ADF file consolidator pipeline with dynamic column headers

Question

I'm trying to do an ADF file consolidator pipeline with dynamic column headers

James Southall 0

I am trying to merge multiple files but lets just say its 2.

File A with 5 columns (Address Line 1, 2, 3, 4, 5)
File B with 7 columns (Address Line 1, 2, 3, 4, 5, Telephone Number, Mobile Number)

I expect the output file to have all 7 columns but those Number fields would be blank for rows that come in from File A.

My current solution I have a schema where I read source files with allow schema drift on, if I only get files that are like File B then all the data is there as expected but if any of the files are like File B the Telephone Number and Mobile Number fields are ALL empty/null even for the rows that had those populated in their initial file.

Nandan Hegde 35,026 Reputation points MVP

2025-05-02T12:56:36.52+00:00

Can you please try the below method

https://learn.microsoft.com/en-us/answers/questions/542994/azure-data-factory-merge-2-csv-files-with-differen

And also share your json code?

James Southall 0

You've linked my question.

What json would you like? The data flow?

{
    "name": "SFTPConsolidation",
    "properties": {
        "type": "MappingDataFlow",
        "typeProperties": {
            "sources": [
                {
                    "dataset": {
                        "referenceName": "pantherManifests",
                        "type": "DatasetReference"
                    },
                    "name": "ReadSourceFiles"
                }
            ],
            "sinks": [
                {
                    "dataset": {
                        "referenceName": "manifests_Processing",
                        "type": "DatasetReference"
                    },
                    "name": "ConsolodateAndOutputFile",
                    "description": "Export data to manifests_Processing",
                    "rejectedDataLinkedService": {
                        "referenceName": "transmissionstorage",
                        "type": "LinkedServiceReference"
                    }
                }
            ],
            "transformations": [
                {
                    "name": "TransformAddress1"
                }
            ],
            "scriptLines": [
                "parameters{",
                "     RunTimeUTC as string,",
                "     outputFilename as string",
                "}",
                "source(output(",
                "          Supplier_Address1 as string,",
                "          Supplier_Address2 as string,",
                "          Supplier_Address3 as string,",
                "          Supplier_Address4 as string,",
                "          Return_Point_Postcode as string,",
                "          Return_Point_Telephone as string",
                "     ),",
                "     allowSchemaDrift: false,",
                "     validateSchema: false,",
                "     ignoreNoFilesFound: true,",
                "     moveFiles: ['in','archive'],",
                "     modifiedBefore: (toTimestamp($RunTimeUTC,'yyyy-MM-dd\\'T\\'HH:mm:ss'))) ~> ReadSourceFiles",
                "ReadSourceFiles derive(Supplier_Address1 = Supplier_Address1+'-'+toString(currentUTC(),'yyyyMMdd')) ~> TransformAddress1",
                "TransformAddress1 sink(allowSchemaDrift: true,",
                "     validateSchema: false,",
                "     partitionFileNames:[($outputFilename)],",
                "     skipDuplicateMapInputs: true,",
                "     skipDuplicateMapOutputs: true,",
                "     quoteAll: true,",
                "     mapColumn(",
                "          Supplier_Address1,",
                "          Supplier_Address2,",
                "          Supplier_Address3,",
                "          Supplier_Address4,",
                "          Return_Point_Postcode,",
                "          Return_Point_Telephone",
                "     ),",
                "     partitionBy('hash', 1)) ~> ConsolodateAndOutputFile"
            ]
        }
    }
}

Nandan Hegde 35,026 Reputation points MVP

2025-05-02T13:14:53.9833333+00:00

Seems to be an issue in mobile via pasting the link.I have edited it. Let me take a look into the JSON as well
James Southall 0 Reputation points

2025-05-02T13:17:55.47+00:00

I looked at the link before creating this question, there seems to be a merging/joining so if rows have the same 5 columns of data then the different fields are merged together on the same row, where as if I have File A with 2 rows and File B with 3 rows, the final file should be 5 rows and row of the rows should have the Telephone data for example.

Your answer

Nandan Hegde 35,026 Reputation points MVP

2025-05-02T12:56:36.52+00:00

Can you please try the below method

https://learn.microsoft.com/en-us/answers/questions/542994/azure-data-factory-merge-2-csv-files-with-differen

And also share your json code?
James Southall 0 Reputation points

2025-05-02T13:12:07.3533333+00:00

You've linked my question.

What json would you like? The data flow?

{ "name": "SFTPConsolidation", "properties": { "type": "MappingDataFlow", "typeProperties": { "sources": [ { "dataset": { "referenceName": "pantherManifests", "type": "DatasetReference" }, "name": "ReadSourceFiles" } ], "sinks": [ { "dataset": { "referenceName": "manifests_Processing", "type": "DatasetReference" }, "name": "ConsolodateAndOutputFile", "description": "Export data to manifests_Processing", "rejectedDataLinkedService": { "referenceName": "transmissionstorage", "type": "LinkedServiceReference" } } ], "transformations": [ { "name": "TransformAddress1" } ], "scriptLines": [ "parameters{", " RunTimeUTC as string,", " outputFilename as string", "}", "source(output(", " Supplier_Address1 as string,", " Supplier_Address2 as string,", " Supplier_Address3 as string,", " Supplier_Address4 as string,", " Return_Point_Postcode as string,", " Return_Point_Telephone as string", " ),", " allowSchemaDrift: false,", " validateSchema: false,", " ignoreNoFilesFound: true,", " moveFiles: ['in','archive'],", " modifiedBefore: (toTimestamp($RunTimeUTC,'yyyy-MM-dd\\'T\\'HH:mm:ss'))) ~> ReadSourceFiles", "ReadSourceFiles derive(Supplier_Address1 = Supplier_Address1+'-'+toString(currentUTC(),'yyyyMMdd')) ~> TransformAddress1", "TransformAddress1 sink(allowSchemaDrift: true,", " validateSchema: false,", " partitionFileNames:[($outputFilename)],", " skipDuplicateMapInputs: true,", " skipDuplicateMapOutputs: true,", " quoteAll: true,", " mapColumn(", " Supplier_Address1,", " Supplier_Address2,", " Supplier_Address3,", " Supplier_Address4,", " Return_Point_Postcode,", " Return_Point_Telephone", " ),", " partitionBy('hash', 1)) ~> ConsolodateAndOutputFile" ] } } }
Nandan Hegde 35,026 Reputation points MVP

2025-05-02T13:14:53.9833333+00:00

Seems to be an issue in mobile via pasting the link.I have edited it. Let me take a look into the JSON as well
James Southall 0 Reputation points

2025-05-02T13:17:55.47+00:00

I looked at the link before creating this question, there seems to be a merging/joining so if rows have the same 5 columns of data then the different fields are merged together on the same row, where as if I have File A with 2 rows and File B with 3 rows, the final file should be 5 rows and row of the rows should have the Telephone data for example.

Share via

I'm trying to do an ADF file consolidator pipeline with dynamic column headers

Your answer