Databricks Autoloader - Column Transformation - Column is not iterable

Question

I am using Azure Databricks Autoloader to process files from ADLS Gen 2 into Delta Lake. I have writen my Foreach batch funtion(pyspark) in the following manner :

#Rename incoming dataframe columns
schemadf = transformschema.renameColumns(microBatchDF,fileconfig)

# Apply simple tranformation on schemadf using createOrReplaceTempView
modifieddf = applytransform(schemadf,targettable,targetdatabase,fileconfig)   

# Add audit cols to modifieddf
transformdf = auditlineage.addauditcols(modifieddf,fileconfig,appid

Code for renameColumns

def renameColumns(dataframe, schema):
  str = schema['Schema']
  splitstr = list(str.split(','))   
  for c,n in zip(dataframe.columns,splitstr):
      dataframe=dataframe.withColumnRenamed(c,n)
  return dataframe

Code for applytransform

def applytransform(inputdf,targettable,targetdatabase,fileconfig):  
  logger.info('Inside applytransform for Database/Table {}.{}',targetdatabase,targettable)
  inputdf.createOrReplaceTempView("src_to_transform")
  lspark = inputdf._jdf.sparkSession()
  if 'TransformQuery' in fileconfig and fileconfig['TransformQuery'] is not None:
    vsqlscript = fileconfig['TransformQuery']
    df = lspark.sql(vsqlscript)    
    logger.info("Applied Tranform")    
    return df
  else:
    logger.info("Passed DF")
    return inputdf

Code for addauditcols

def addauditcols(inputdf,fileconfig,app_id):
    now = datetime.datetime.now()
    print(type(inputdf))    
    createdby = 'DatabricksJob-'+app_id
    datasource = fileconfig['Datasource']
    recordactiveind = 'Y'
    df = inputdf.withColumn('datasource',lit(datasource)).\
    withColumn('createdtimestamp',lit(now)).\
    withColumn('lastmodifiedtimestamp',lit(now)).\
    withColumn('createduserid',lit(createdby)).\
    withColumn('lastmodifieduserid',lit(createdby)).\
    withColumn('filepath',input_file_name()).\
    withColumn('recordactiveind',lit(recordactiveind))
    return df

applytransform module returns a py4j.java_gateway.JavaObject instead of a regular pyspark.sql.dataframe.DataFrame and hence I am not able to perform simple withColumn() type transformations on modifieddf within addauditcols module

The error I get is as follows:

2021-12-05 21:09:57.274 | INFO     | __main__:main:73 - modifieddf Type::: 

2021-12-05 21:09:57.421 | ERROR    | __main__:main:91 - Operating Failed for md_customer, with Exception Column is not iterable
Traceback (most recent call last):

  File "c:/Users/asdsad/integration-app\load2cleansed.py", line 99, in 
    main()
    └ 

> File "c:/Users/asdsad/integration-app\load2cleansed.py", line 76, in main
    transformdf = auditlineage.addauditcols(modifieddf,fileconfig,appid)
                  │            │            │          │          └ 'local-1638760184357'
                  │            │            │          └ {'Schema': 'customernumber,customername,addrln1,city,statename,statecode,postalcode,countrycode,activeflag,sourcelastmodified...
                  │            │            └ JavaObject id=o48
                  │            └ 
                  └ 
         └ JavaObject id=o48

  File "C:\Users	estapp\lib\site-packages\py4j\java_gateway.py", line 1296, in __call__
    args_command, temp_args = self._build_args(*args)
                              │    │            └ ('datasource', Column<'DUMMY-CUST'>)
                              │    └ 
                              └ 

  File "C:\Users	estapp\lib\site-packages\py4j\java_gateway.py", line 1260, in _build_args
    (new_args, temp_args) = self._get_args(args)
                            │    │         └ ('datasource', Column<'DUMMY-CUST'>)
                            │    └ 
                            └ 

  File "C:\Users	estapp\lib\site-packages\py4j\java_gateway.py", line 1247, in _get_args
    temp_arg = converter.convert(arg, self.gateway_client)
               │         │       │    │    └ 
               │         │       │    └ 
               │         │       └ Column<'DUMMY-CUST'>
               │         └ 
               └ 

  File "C:\Users	estapp\lib\site-packages\py4j\java_collections.py", line 510, in convert
    for element in object:
                   └ Column<'DUMMY-CUST'>

  File "C:\Users	estapp\lib\site-packages\pyspark\sql\column.py", line 470, in __iter__
    raise TypeError("Column is not iterable")

TypeError: Column is not iterable

Any help is appreciated

user_10972959 · Answer

Please remove lspark = inputdf._jdf.sparkSession()

it is used to sql upsert command to delta like merge not to return dataframe.

Please just use spark.sql(vsqlscript)

If it doesn't help please share your vsqlscript code also.

user_10180813 · Answer

Thanks. Instead of createOrReplaceTempView , i updated to createOrReplaceGlobalTempView and updated the vsqlscript to read from select * from global_temp.src_to_transform Made the following change in applytransform:

def applytransform(inputdf,targettable,targetdatabase,fileconfig):
  
  logger.info('Inside applytransform for Database/Table {}.{}',targetdatabase,targettable)
  
  **## Store in global Temp databricks database**
  inputdf.createOrReplaceGlobalTempView("src_to_transform")
  lspark = inputdf._jdf.sparkSession()
  if 'TransformQuery' in fileconfig and fileconfig['TransformQuery'] is not None:
    vsqlscript = fileconfig['TransformQuery']
    #df = lspark.sql(vsqlscript)
    df = spark.sql(vsqlscript)    
    logger.info("Applied Tranform")
    df.show()
    return df
  else:
    logger.info("Passed DF")
    return inputdf

Databricks Autoloader - Column Transformation - Column is not iterable

2 Answers