// input
{
  "a": {
     "b": 1
  }
}

Python: events.select("a.b")
 Scala: events.select("a.b")
   SQL: select a.b from events

// output
{
  "b": 1
}

// input
{
  "a": {
     "b": 1,
     "c": 2
  }
}

Python:  events.select("a.*")
 Scala:  events.select("a.*")
   SQL:  select a.* from events

// output
{
  "b": 1,
  "c": 2
}

// input
{
  "a": 1,
  "b": 2,
  "c": 3
}

Python: events.select(struct(col("a").alias("y")).alias("x"))
 Scala: events.select(struct('a as 'y) as 'x)
   SQL: select named_struct("y", a) as x from events

// output
{
  "x": {
    "y": 1
  }
}

// input
{
  "a": 1,
  "b": 2
}

Python: events.select(struct("*").alias("x"))
 Scala: events.select(struct("*") as 'x)
   SQL: select struct(*) as x from events

// output
{
  "x": {
    "a": 1,
    "b": 2
  }
}

// input
{
  "a": [1, 2]
}

Python: events.select(col("a").getItem(0).alias("x"))
 Scala: events.select('a.getItem(0) as 'x)
   SQL: select a[0] as x from events

// output
{ "x": 1 }

// input
{
  "a": {
    "b": 1
  }
}

Python: events.select(col("a").getItem("b").alias("x"))
 Scala: events.select('a.getItem("b") as 'x)
   SQL: select a['b'] as x from events

// output
{ "x": 1 }

// input
{
  "a": [1, 2]
}

Python: events.select(explode("a").alias("x"))
 Scala: events.select(explode('a) as 'x)
   SQL: select explode(a) as x from events

// output
[{ "x": 1 }, { "x": 2 }]

// input
{
  "a": {
    "b": 1,
    "c": 2
  }
}

Python: events.select(explode("a").alias("x", "y"))
 Scala: events.select(explode('a) as Seq("x", "y"))
   SQL: select explode(a) as (x, y) from events

// output
[{ "x": "b", "y": 1 }, { "x": "c", "y": 2 }]

collect_list()和collect_set()可用于聚集物品进入阵列。

// input
[{ "x": 1 }, { "x": 2 }]

Python: events.select(collect_list("x").alias("x"))
 Scala: events.select(collect_list('x) as 'x)
   SQL: select collect_list(x) as x from events

// output
{ "x": [1, 2] }

// input
[{ "x": 1, "y": "a" }, { "x": 2, "y": "b" }]

Python: events.groupBy("y").agg(collect_list("x").alias("x"))
 Scala: events.groupBy("y").agg(collect_list('x) as 'x)
   SQL: select y, collect_list(x) as x from events group by y

// output
[{ "y": "a", "x": [1]}, { "y": "b", "x": [2]}]

从数组中的每个项目中选择一个字段

当您在数组中使用点符号时，我们返回一个新数组，该数组已从每个数组元素中选择该字段。

// input
{
  "a": [
    {"b": 1},
    {"b": 2}
  ]
}

Python: events.select("a.b")
 Scala: events.select("a.b")
   SQL: select a.b from events

// output
{
  "b": [1, 2]
}

to_json（）和from_json（）的权力

如果您真的想保留列的复杂结构，但是您需要将其编码为字符串来存储？你注定了吗 当然不是！Spark SQL提供to_json()了将结构体编码为字符串并将from_json()结构检索为复杂类型的函数。使用JSON字符串作为列在读取或写入流媒体源（如Kafka）时非常有用。每个Kafka键值记录将被添加一些元数据，如摄入时间戳到Kafka，Kafka中的偏移量等。如果包含数据的“值”字段是JSON，您可以使用它from_json()来提取数据，丰富它，清理它，然后将其下游推送到卡夫卡再次或将其写入文件。

将结构编码为json

to_json()可以将结构体转换成JSON字符串。当您将数据写入Kafka时，当您想要将多个列重新编码为一个列时，此方法特别有用。此方法目前在SQL中不可用。

// input
{
  "a": {
    "b": 1
  }
}

Python: events.select(to_json("a").alias("c"))
 Scala: events.select(to_json('a) as 'c)

// output
{
  "c": "{\"b\":1}"
}

将json列解码为结构体

from_json()可用于将具有JSON数据的字符串列转换为结构体。那么你可以按照上述的方式平坦化结构体，使其具有单独的列。此方法目前在SQL中不可用。

// input
{
  "a": "{\"b\":1}"
}

Python: 
  schema = StructType().add("b", IntegerType())
  events.select(from_json("a", schema).alias("c"))
Scala:
  val schema = new StructType().add("b", IntegerType)
  events.select(from_json('a, schema) as 'c)

// output
{
  "c": {
    "b": 1
  }
}

有时您可能希望将JSON字符串的一部分仍保留为JSON，以避免模式中的复杂性过高。

// input
{
  "a": "{\"b\":{\"x\":1,\"y\":{\"z\":2}}}"
}

Python: 
  schema = StructType().add("b", StructType().add("x", IntegerType())
                              .add("y", StringType()))
  events.select(from_json("a", schema).alias("c"))
Scala:
  val schema = new StructType().add("b", new StructType().add("x", IntegerType)
    .add("y", StringType))
  events.select(from_json('a, schema) as 'c)

// output
{
  "c": {
    "b": {
      "x": 1,
      "y": "{\"z\":2}"
    }
  }
}

从包含JSON的列中解析一组字段

json_tuple() 可以使用JSON数据来提取字符串列中可用的字段。

// input
{
  "a": "{\"b\":1}"
}

Python: events.select(json_tuple("a", "b").alias("c"))
Scala:  events.select(json_tuple('a, "b") as 'c)
SQL:    select json_tuple(a, "b") as c from events

// output
{ "c": 1 }

spark 结构化流中的应用：
输入是一串json字符串在kafka = $"value", 
{ 下边为提取一个元素，和提取多个元素的方法 }
var streamingSelectDF = streamingInputDF.select(get_json_object(($"value").cast("string"), "$.zip").alias("zip"))
//: org.apache.spark.sql.DataFrame = [zip: string]
var streamingSelectDF = streamingInputDF.select(json_tuple(($"value").cast("string"), "$.zip", "$.isp"))//: org.apache.spark.sql.DataFrame = [c0: string, c1: string]

有时一个字符串列可能不是像JSON一样自我描述，但仍可能有一个格式正确的结构。例如，它可以是使用特定的Log4j格式生成的日志消息。Spark SQL可以轻松地为您构建这些字符串！

解析一个格式良好的字符串列

regexp_extract() 可用于使用正则表达式解析字符串。

// input
[{ "a": "x: 1" }, { "a": "y: 2" }]

Python: events.select(regexp_extract("a", "([a-z]):", 1).alias("c"))
Scala:  events.select(regexp_extract('a, "([a-z]):", 1) as 'c)
SQL:    select regexp_extract(a, "([a-z]):", 1) as c from events

// output
[{ "c": "x" }, { "c": "y" }]

这是很多变化！现在我们来看一下现实生活中的一些用例，把所有这些数据格式和数据处理能力都用到很好的用处。

利用这一切权力

在Databricks，我们从我们的服务收集日志，并在客户受到影响之前使用它们执行实时监控以检测问题。日志文件是非结构化文件，但它们是可解析的，因为它们具有明确定义的Log4j格式。我们运行日志收集器服务，将每个日志条目和关于JSON的条目（例如源）的附加元数据发送到Kinesis。这些JSON记录然后批量上传到S3作为文件。查询这些JSON日志以回答任何问题是乏味的：这些文件包含重复项，并且对于回答任何查询，即使它涉及单个列，整个JSON记录也许需要反序列化。

为了解决这个问题，我们运行一个读取这些JSON记录并对元数据执行重复数据删除的管道。现在我们留下原始的日志记录，可能是JSON格式或非结构化文本。如果我们正在处理JSON，我们使用from_json()上面描述的几个转换来格式化数据。如果是文本，我们使用诸如regexp_extract()将Log4j格式解析成更结构化的方法。一旦完成了我们所有的变革和重组，我们将记录按日期保存在Parquet中。当我们回答“我们在10：00-10：30这个特定的服务之间看到多少错误信息”这样的问题时，这给我们10-100倍的加速？加速可归因于：

• 我们不再支付反序列化JSON记录的价格
• 我们不必对原始日志消息执行复杂的字符串比较
• 我们只需要在我们的查询中提取两列：时间和日志级别

以下是我们在客户中看到的一些更常见的用例：

“我想用我的数据运行机器学习管道。我的数据已经被预处理，我将使用我所有的功能。

当您访问整行数据时，Avro是一个不错的选择。

“我有一个IoT用例，我的传感器给我发送事件。对于每个事件，重要的元数据是不同的。“

如果您希望在架构中具有灵活性，则可以考虑使用JSON来存储数据。

“我想在报纸文章或情感分析上对产品评论进行语音识别算法。”

如果您的数据可能没有固定的模式，也没有固定的模式/结构，那么将其存储为纯文本文件可能会更容易。您还可以拥有一个管道，可对此非结构化数据执行特征提取，并将其存储为Avro，以准备您的机器学习流程。

结论

在这篇博文中，我们讨论了Spark SQL如何允许您从许多来源和格式中使用数据，并轻松执行这些数据格式之间的转换和交换。我们分享了我们如何在Databricks处理我们的数据，并考虑了其他生产用例，您可能想要做不同的事情。

Spark SQL为您提供必要的工具，以便以任何形式访问您的数据，无论其格式如何，并为下游应用程序做准备，无论是在流数据上具有低延迟，还是对旧数据进行高吞吐量！

在本系列的未来博客文章中，我们将介绍以下内容：

• 监控流媒体应用程序
• 将结构化流与Apache Kafka集成
• 使用结构化流计算事件时间聚合

如果您想了解有关结构化流的更多信息，以下是一些有用的链接。

• 以前的博客帖子解释了结构化流的动机和概念：
  • 持续应用：在Apache Spark 2.0中不断发展
  • Apache Spark中的结构化流式传输
• 在Apache Spark 2.2中使用结构化流处理Apache Kafka中的数据
  • 在Apache Spark中实时流式传输ETL结构化流2.1
• 结构化流编程指南
• 在2017年Spark Summit峰会上谈谈 - 制作结构化流媒体，准备制作和未来的方向

最后，尝试我们的示例笔记本，演示如何在Databricks中的Python，Scala或SQL中转换复杂数据类型：

• Python笔记本
• Scala笔记本
• SQL笔记本