Spark DataFame : JDBC Write Auto generated fields

user_4945499 0
Viewed 971

I have dataframe that matches exactly like table except for primary key which is auto generated .

Below is my table

    ID         |FirstName |LastName    |CreOn      |CreBy
auto-generated |Varchar(20)|Varchar(20)| timestamp| Varchar(20)

Below is my dataframe

FirstName|LastName|CreOn    |CreBy
String   |String  |timestamp| String

When I use spark.jdbc.write with Mode 'append', I get below error

Permission denied for schema abc Position 14

How do we handle Autogenerated Fields.

Should I prefer using Python to do the JDBC operation as compared with Pyspark as I will have greater control over batch size and rollback

pyspark apache-spark postgresql python
2 Answers

It depends upon your data size....if the number of rows is huge python will most probably fail and you won't be able to load data.

One possible way to do this is use staging table. Load data from spark to your staging table. Then run a query to move data from staging to your main table using python libraries.

As far as batchsize is concerned, you can configure it in spark jdbc write also.

For really huge data I would suggest use SQOOP to move data as spark cannot form concurrent connections with db while sqoop can have establish multiple mapper to write data to your db.

user_53527480

When a dataframe is saved into a SQL database, Spark will only writes into those columns in the database that are present in the dataframe. So if the ID column is not part of the dataframe, Spark will ignore it during the insert operation.

The insert statement is created in JdbcUtils.getInsertStatement(), and here only columns that are part of the underlying rdd are included in the create statement:

val columns = 
      [...]
      rddSchema.fields.map { col =>
        val normalizedName = tableColumnNames.find(f => columnNameEquality(f, col.name)).getOrElse {
          throw new AnalysisException(s"""Column "${col.name}" not found in schema $tableSchema""")
        }
        dialect.quoteIdentifier(normalizedName)
      }.mkString(",")
      [...]
s"INSERT INTO $table ($columns) VALUES ($placeholders)"

For example given the table definition

create table address (
    id serial, 
    FirstName varchar(20), 
    LastName varchar(20),
    CreOn timestamp,
    CreBy varchar(20),
    constraint pk primary key (id))

and the Python code

df = spark.createDataFrame(
    [("John", "Doe", "1970-01-02 03:46:40", "py2")],
    ['FirstName','LastName','CreOn','CreBy']
)

df.write.mode("append").jdbc(<jdbc url>, "address",  \
                             properties={"driver":...,"user": ..., "password": ...})

Spark creates the insert statement

INSERT INTO address ("firstname","lastname","creon","creby") VALUES (?,?,?,?)

and the insert operation is successful.

So autogenerated fields should simply not be part of the dataframe and the error Permission denied is probably not related to the autogenerated field.

user_21298010
Related