Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

支持的数据类型

Spark 连接器支持 Amazon Redshift 中的以下数据类型。有关 Amazon Redshift 中支持的数据类型的完整列表，请参阅数据类型。如果某个数据类型不在下表中，则 Spark 连接器不支持该数据类型。

复杂的数据类型

您可以使用 spark 连接器在 Redshift SUPER 数据类型列中读写 Spark 复杂数据类型，如 ArrayType、MapType 和 StructType。如果您在读取操作期间提供架构，则该列中的数据将在 Spark 中转换为相应的复杂类型，包括任何嵌套类型。此外，如果启用 autopushdown，嵌套属性、映射值和数组索引的投影将下推到 Redshift，这样，当只访问一部分数据时，就不再需要卸载整个嵌套数据结构。

从连接器写入 DataFrame 时，任何类型为 MapType（使用 StringType）、StructType 或 ArrayType 的列都会写入 Redshift SUPER 数据类型列。在写入这些嵌套数据结构时，tempformat 参数必须为类型 CSV、CSV GZIP 或 PARQUET。使用 AVRO 将导致异常。写入一个键类型不是 StringType 的 MapType 数据结构也会导致异常。

StructType

以下示例演示如何使用包含结构的 SUPER 数据类型创建表

create table contains_super (a super);

然后，您可以使用连接器，使用下面示例中的类似架构，从表中的 SUPER 列 a 查询 StringType 字段 hello。

import org.apache.spark.sql.types._

val sc = // existing SparkContext
val sqlContext = new SQLContext(sc)

val schema = StructType(StructField("a", StructType(StructField("hello", StringType) ::Nil)) :: Nil)

val helloDF = sqlContext.read
.format("io.github.spark_redshift_community.spark.redshift")
.option("url", jdbcURL )
.option("tempdir", tempS3Dir)
.option("dbtable", "contains_super")
.schema(schema)
.load().selectExpr("a.hello")

以下示例演示如何向列 a 写入结构。

import org.apache.spark.sql.types._
import org.apache.spark.sql._

val sc = // existing SparkContext
val sqlContext = new SQLContext(sc)

val schema = StructType(StructField("a", StructType(StructField("hello", StringType) ::Nil)) :: Nil)
val data = sc.parallelize(Seq(Row(Row("world"))))
val mydf = sqlContext.createDataFrame(data, schema)

mydf.write.format("io.github.spark_redshift_community.spark.redshift").
option("url", jdbcUrl).
option("dbtable", tableName).
option("tempdir", tempS3Dir).
option("tempformat", "CSV").
mode(SaveMode.Append).save

MapType

如果您更喜欢使用 MapType 来表示数据，那么您可以在架构中使用 MapType 数据结构，并检索映射中与键对应的值。请注意，MapType 数据结构中的所有键都必须是 String 类型，并且所有值都必须是相同的类型，例如 int。

以下示例演示如何获取列 a 中键 hello 的值。

import org.apache.spark.sql.types._

val sc = // existing SparkContext
val sqlContext = new SQLContext(sc)

val schema = StructType(StructField("a", MapType(StringType, IntegerType))::Nil)

val helloDF = sqlContext.read
    .format("io.github.spark_redshift_community.spark.redshift")
    .option("url", jdbcURL )
    .option("tempdir", tempS3Dir)
    .option("dbtable", "contains_super")
    .schema(schema)
    .load().selectExpr("a['hello']")

ArrayType

如果该列包含数组而不是结构，则可以使用连接器查询数组中的第一个元素。

import org.apache.spark.sql.types._

val sc = // existing SparkContext
val sqlContext = new SQLContext(sc)

val schema = StructType(StructField("a", ArrayType(IntegerType)):: Nil)

val helloDF = sqlContext.read
    .format("io.github.spark_redshift_community.spark.redshift")
    .option("url", jdbcURL )
    .option("tempdir", tempS3Dir)
    .option("dbtable", "contains_super")
    .schema(schema)
    .load().selectExpr("a[0]")

限制

通过 spark 连接器使用复杂数据类型有以下限制：