将数据从 PySpark 加载到 Redshift 时如何执行列编码

Posted 2023-03-31

【中文标题】将数据从 PySpark 加载到 Redshift 时如何执行列编码

【英文标题】：How to perform column encoding when loading data from PySpark to Redshift

【发布时间】：2016-07-22 18:57:13

【问题描述】：

我正在尝试使用 pyspark 直接将 S3 上以镶木地板格式的数据加载到 aws redshift。我能够做到这一点，但是当我看到表定义中列的编码时，它是一致的。我想让它保持一致，特别是我希望他们所有人都是 lzo。以下是来自单个表的不一致的数据类型列表。

+-------------------------------+-------------------+
|    data_type                  |  encoding         |
+-------------------------------+-------------------+
| bigint                        | delta             |
| bigint                        | delta32k          | 
| character varying(256)        | lzo               |
| bigint                        | runlength         |
| bigint                        | bytedict          |
| timestamp without time zone   | bytedict          |
| integer                       | runlength         |
+-------------------------------+-------------------+

有人可以帮助我如何在 pyspark 中执行此操作。我在 com.databricks:spark-redshift_2.10:1.0.0

中看不到任何列编码选项

 x.write.format("com.databricks.spark.redshift")
.option("url","jdbc:redshift://<url>:<port>/<schema>?user=<user>&password=<pass>")
.option("dbtable","<tbl_nm>")
.option("diststyle","KEY").option("distkey","<key>")
.option("sortkeyspec","SORTKEY(<sort1>)")
.option("tempdir","<path>")
.mode("error").save()

【参考方案1】：

我在PR 178 中找到了指定列编码的相关位。

因此，您无需通过 .read.option('encoding', 'lzo') 之类的方式指定编码。您需要使用元数据创建一个模式对象，该对象在创建数据帧时指定编码。以 Python 为例：

%pyspark 

from pyspark.sql.types import IntegerType, StringType, StructType, StructField

metadata = 'encoding':'LZO'

schema = StructType([
    StructField("id", IntegerType(), True, metadata),
    StructField("name", StringType(), True, metadata)])

df = spark.createDataFrame([(1, 'Alice')], schema)

df.write \
  .format("com.databricks.spark.redshift") \
  .option("url", "jdbc:redshift://example.com:5439/db_foo?user=user_bar&password=pass_baz") \
  .option("dbtable", "foo") \
  .option("tempdir", "s3a://foo/bar") \
  .mode("error") \
  .save()

验证：

select "column", "encoding" from pg_table_def where tablename = 'foo';
 column | encoding
--------+----------
 id     | lzo
 name   | lzo
(2 rows)