Dávám přednost použití nativního datového rámce Spark, protože mi to umožňuje více přizpůsobení. Mohu použít stringtype vlastnost k přetypování pole json z datového rámce do pole jsonb v tabulce. V tomto případě má můj datový rámec dvě pole.
from pyspark import SparkConf
sc = SparkContext.getOrCreate(SparkConf())
spark = SparkSession(sc)
df = spark.read.format('csv') \
.option('delimiter','|') \
.option('header','True') \
.load('your_path')
##some transformation...
url = 'jdbc:postgresql://your_host:5432/your_databasename'
properties = {'user':'*****',
'password':'*****',
'driver': "org.postgresql.Driver",
'stringtype':"unspecified"}
df.write.jdbc(url=url, table='your_tablename', mode='append', properties=properties)
Před spuštěním výše uvedeného skriptu byste měli vytvořit tabulku v postgresql, protože vlastnost mode je nastaveno jako připojit . Takto:
create table your_tablename
(
my_json_field jsonb,
another_field int
)