在 R 中使用 SparklyR 更改嵌套列名称答案

【问题标题】：Changing nested column names using SparklyR in R在 R 中使用 SparklyR 更改嵌套列名称
【发布时间】：2019-02-15 05:52:51
【问题描述】：

我已经参考了这里提到的所有链接：

1) Link-1 2) Link-2 3) Link-3 4) Link-4

以下 R 代码是使用 Sparklyr 包编写的。它读取巨大的 JSON 文件并创建数据库模式。

sc <- spark_connect(master = "local", config = conf, version = '2.2.0') # Connection
sample_tbl <- spark_read_json(sc,name="example",path="example.json", header = TRUE, 
                              memory = FALSE, overwrite = TRUE) # reads JSON file
sample_tbl <- sdf_schema_viewer(sample_tbl) # to create db schema
df <- tbl(sc,"example") # to create lookup table

它创建了以下数据库模式

现在，

如果我重命名第一级列，那么它可以工作。

例如，

df %>% rename(ent = entities)

但是当我运行第二个深层嵌套列时，它不会重命名。

df %>% rename(e_hashtags = entities.hashtags)

显示错误：

Error in .f(.x[[i]], ...) : object 'entities.hashtags' not found

问题

我的问题是，如何将第 3 层嵌套列重命名为第 4 层嵌套列？

请参考上述数据库架构。

【问题讨论】：

标签： r apache-spark sparklyr

【解决方案1】：

Spark 本身不支持重命名单个嵌套字段。您必须铸造或重建整个结构。为简单起见，我们假设数据如下所示：

cat('{"contributors": "foo", "coordinates": "bar", "entities": {"hashtags": ["foo", "bar"], "media": "missing"}}',  file = "/tmp/example.json")
df <- spark_read_json(sc, "df", "/tmp/example.json", overwrite=TRUE)

df %>% spark_dataframe() %>% invoke("schema") %>% invoke("treeString") %>% cat()

root
 |-- contributors: string (nullable = true)
 |-- coordinates: string (nullable = true)
 |-- entities: struct (nullable = true)
 |    |-- hashtags: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- media: string (nullable = true)

用简单的字符串表示：

df %>% 
  spark_dataframe() %>% 
  invoke("schema") %>% 
  invoke("simpleString") %>% 
  cat(sep = "\n")

struct<contributors:string,coordinates:string,entities:struct<hashtags:array<string>,media:string>>

通过强制转换，您必须使用匹配类型描述来定义表达式：

expr_cast <- invoke_static(
  sc, "org.apache.spark.sql.functions", "expr",
  "CAST(entities AS struct<e_hashtags:array<string>,media:string>)"
)

df_cast <- df %>% 
  spark_dataframe() %>% 
  invoke("withColumn", "entities", expr_cast) %>% 
  sdf_register()

df_cast %>% spark_dataframe() %>% invoke("schema") %>% invoke("treeString") %>% cat()

root
 |-- contributors: string (nullable = true)
 |-- coordinates: string (nullable = true)
 |-- entities: struct (nullable = true)
 |    |-- e_hashtags: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- media: string (nullable = true)

要重建结构，您必须匹配所有组件：

expr_struct <- invoke_static(
  sc, "org.apache.spark.sql.functions", "expr",
  "struct(entities.hashtags AS e_hashtags, entities.media)"
)

df_struct <- df %>% 
  spark_dataframe() %>% 
  invoke("withColumn", "entities", expr_struct) %>% 
  sdf_register()

df_struct %>% spark_dataframe() %>% invoke("schema") %>% invoke("treeString") %>% cat()

root
 |-- contributors: string (nullable = true)
 |-- coordinates: string (nullable = true)
 |-- entities: struct (nullable = false)
 |    |-- e_hashtags: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- media: string (nullable = true)

【讨论】：

这不会以任何实际方式影响此答案。列始终被标识，但它的完全限定名称。歧义仅在极少数情况下发生，即您在同一路径上具有相同名称的列（假设您有entities.url 和entities.url - 虽然这是边界情况，通常表明管道或严重数据中的错误问题）。
你必须重新创建一个完整的结构。如果您有多个嵌套级别，则需要多个嵌套结构来反映 struct(..., struct(...)) 等等。然而，并不是所有的结构都可以这样建造。