Загрузите данные jsonb из postgresql в pyspark и сохраните их в MapType
Таблицы продуктов создают и вставляют скрипты:
create table products (product_id varchar, description varchar, attributes jsonb, tax_rate decimal);
insert into products values ('P1', 'Detergent', '{"cost": 45.50, "size": "10g"}', 5.0 );
insert into products values ('P2', 'Bread', '{"cost": 45.5, "size": "200g"}',3.5);
Я пытаюсь сохранить данные jsonb из postgresql в формат данных/словаря типа карты в PySpark, а затем извлечь «стоимость» и «размер» из столбца «атрибуты» в отдельные столбцы. Но PySpark читает данные jsonb как строку.
Код PySpark для чтения данных из Postgresql:
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, MapType, DecimalType
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.jars", "C:\\Users\\nupsingh\\Documents\\Jars\\postgresql-42.7.3.jar") \
.getOrCreate()
schema = StructType([
StructField('product_id',
StringType(), True),
StructField('description',
StringType(), True),
StructField('attributes',
MapType(StringType(),IntegerType()),False),
StructField('tax_rate',
DecimalType(), True)
])
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://localhost:5432/postgres") \
.option("dbtable", "products") \
.option("user", "user1") \
.option("password", "password") \
.option("driver", "org.postgresql.Driver") \
.option("schema", schema) \
.load()
df.show()
df.printSchema()
attributes_col = df.select("attributes")
attributes_col.show()
products_df = attributes_col.withColumn("cost",
col("attributes")["cost"]).withColumn(
"size", col("attributes")["size"])
products_df.show()
Ответы 1
Короткий ответ: вы не можете напрямую загрузить поле jsonb в MapType в Pyspark.
Вот почему:
Когда вы считываете данные из базы данных с помощью источника данных JDBC PySpark, PySpark полагается на метаданные, предоставляемые драйвером JDBC, для определения типов данных столбцов. Драйвер JDBC сопоставляет типы данных базы данных с типами данных Java, которые затем сопоставляются с типами данных PySpark. В случае типа данных Postgres jsonb драйвер JDBC postgres сопоставляет его с типом Java String. PySpark при взаимодействии с драйвером JDBC получает данные jsonb в виде строки и выводит их как StringType в схеме. PySpark не имеет встроенного понимания конкретной структуры данных jsonb.
Есть ли обходной путь?
Да, вы можете достичь своей цели в два этапа: 1- прочитать данные jsonb как строку и 2- использовать from_json, чтобы преобразовать их в MapType. Вот как вы можете это сделать:
from pyspark.sql.types import StructType, StructField, StringType, MapType, IntegerType, DecimalType
from pyspark.sql.functions import from_json, col
json_schema = MapType(StringType(),IntegerType()))
schema = StructType([
StructField('product_id', StringType(), True),
StructField('description', StringType(), True),
StructField('attributes', MapType(StringType(), IntegerType(), True), True),
StructField('tax_rate', DecimalType(), True),
StructField('attributes_json', StringType(), True) # Add a temporary field to hold the jsonb data
])
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://localhost:5432/postgres") \
.option("dbtable", "products") \
.option("user", "user1") \
.option("password", "password") \
.option("driver", "org.postgresql.Driver") \
.option("schema", schema) \
.load()
# Convert the jsonb column to a MapType
df = df.withColumn("attributes", from_json(col("attributes_json"), json_schema)) \
.drop("attributes_json")
@Nupur Просто поменяйте местами attributes_json и atrributes. Я использовал другое имя, чтобы продемонстрировать двухэтапный процесс.
Да, я только что понял. Большое спасибо!
Спасибо за помощь. Когда я запускаю этот код, я получаю сообщение об ошибке: pyspark.errors.Exceptions.captured.AnalysisException: [UNRESOLVED_COLUMN.WITH_SUGGESTION] Невозможно разрешить параметр столбца или функции с именем
attributes_json. Вы имели в виду одно из следующих? [attributes,product_id,tax_rate,description].; 'Проект [product_id#0, описание#1, from_json(MapType(StringType,IntegerType,true), 'attributes_json, Some(Asia/Calcutta)) AS атрибуты#25, Tax_rate#3] +- Отношение [product_id#0,description #1,атрибуты#2,tax_rate#3] JDBCRelation(продукты) [numPartitions=1]