Dynamic Schemas mapping to postgresSQL sink in ADF Dataflow

Question

Hi,
We have csv files with dynamic schema coming from customers to our system through datalake.
I am able to pick it up one by one and do transformation and bring it to a format as below

customer1
tenantid,accountnumber,accountname,postcode
"111","A1234","A good name","AA4 3BZ"
"111","A1234","Another good name","AB4 3BZ"
...

customer2
tenantid,accountnumber,postcode
"112","B1234","AA4 3BZ"
"112","B1234","AB4 3BZ"
...

They must be mapped to the below postgresql table

tenantid, accountnumber, accountname, postcode

My dataflow is as follows:

parameters{
    TenantId as integer
}
source(allowSchemaDrift: true,
    validateSchema: false,
    inferDriftedColumnTypes: true,
    ignoreNoFilesFound: false) ~> CuratedZoneData
CuratedZoneData alterRow(insertIf(true()),
    updateIf(true())) ~> AlterRow2
AlterRow2 derive(each(match(type == 'string'), $$ = $$)) ~> DerivedColumn3
DerivedColumn3 sink(allowSchemaDrift: true,
    validateSchema: false,
    input(
        accountnumber as string,
        accountname as string,
        tenantid as string,
        postcode as string
    ),
    deletable:false,
    insertable:true,
    updateable:false,
    upsertable:false,
    format: 'table',
    skipDuplicateMapInputs: true,
    skipDuplicateMapOutputs: true,
    saveOrder: 1) ~> sink1

This results in a blank table and nothing is inserted or updated.

What I expect is for data from both the csv's to be inserted/upserted to the database. Note that all the columns are nullable in the database and all the columns are string.

Any help in identifying where I am going wrong is appreciated.

Answer 1

Take out the Alter Row and just insert the rows. Make sure your sink settings allows inserts. Also take out your derived column transformation. Use auto-mapping in your sink.