PySpark sql запись dataFrame со столбцом json в mysql JDBC - PullRequest
Новая
       36

PySpark sql запись dataFrame со столбцом json в mysql JDBC

0 голосов
/ 27 февраля 2020

Я пытаюсь массово записать DataFrame в базу данных mysql JDB C. Я использую databricks / pyspark. sql для записи DataFrames в таблицу. В этой таблице есть столбец, который принимает данные json (двоичные данные). Я преобразовал json объект в StructType со следующей структурой:

json структура объекта и преобразование в фрейм данных: 

schema_dict = {'fields': [
    {'metadata': {}, 'name': 'dict', 'nullable': True, 'type': {"containsNull": True, "elementType":{'fields': [
      {'metadata': {}, 'name': 'y1', 'nullable': True, 'type': 'integer'},
      {'metadata': {}, 'name': 'y2', 'nullable': True, 'type': 'integer'}
    ],"type": 'struct'}, "type": 'array'}}
], 'type': 'struct'}

cSchema = StructType([StructField("x1", IntegerType()),StructField("x2", IntegerType()),StructField("x3", IntegerType()),StructField("x4", TimestampType()), StructField("x5", IntegerType()), StructField("x6", IntegerType()),
                     StructField("x7", IntegerType()), StructField("x8", TimestampType()), StructField("x9", IntegerType()), StructField("x10", StructType.fromJson(schema_dict))])
df = spark.createDataFrame(parsedList,schema=cSchema)

Выходной фрейм данных: 

df:pyspark.sql.dataframe.DataFrame
x1:integer
x2:integer
x3:integer
x4:timestamp
x5:integer
x6:integer
x7:integer
x8:timestamp
x9:integer
x10:struct
    dict:array
        element:struct
              y1:integer
              y2:integer

Теперь я пытаюсь записать этот кадр данных в таблицу mysql, используя таблицу mysql. 

import urllib 
from pyspark.sql import SQLContext
from pyspark.sql.functions import regexp_replace, col
sqlContext = SQLContext(sc)
sqlContext

driver = "org.mariadb.jdbc.Driver"
url = "jdbc:mysql://dburl?rewriteBatchedStatements=true"
trial = "dbname.tablename"
user = "dbuser"
password = "dbpassword"
properties = {
    "user": user,
    "password": password,
    "driver": driver
}
df.write.jdbc(url=url, table=trial, mode="append", properties = properties)

Я получаю эту ошибку: 

An error occurred while calling o2118.jdbc.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 15 in stage 176.0 failed 4 times, most recent failure: Lost task 15.3 in stage 176.0 (TID 9528, 10.168.231.82, executor 5): java.lang.IllegalArgumentException: Can't get JDBC type for struct<dict:array<struct<y1:int,y2:int>>>

Любые идеи по Как записать фрейм данных, который имеет столбец json в таблицу mysql? или как решить эту проблему?

Я использую Databricks 5.5 LTS (включая Apache Spark 2.4.3, Scala 2.11)

...