Как сгладить данные разных типов данных с помощью пакета Sparklyr?

Вступление

Код R написан с использованием пакета Sparklyr для создания схемы базы данных. [Воспроизводится код и база данных]

Существующий результат

root
|-- contributors : string
|-- created_at : string
|-- entities (struct)
|     |-- hashtags (array) : [string]
|     |-- media (array)
|     |     |-- additional_media_info (struct)
|     |     |       |-- description : string
|     |     |       |-- embeddable : boolean
|     |     |       |-- monetizable : bollean
|     |     |-- diplay_url : string
|     |     |-- id : long
|     |     |-- id_str : string
|     |-- urls (array)     
|-- extended_entities (struct)
|-- retweeted_status (struct)
|-- user (struct)

Я хочу сгладить эту структуру, как показано ниже,

ожидаемый результат

root
|-- contributors : string
|-- created_at : string
|-- entities (struct)
|-- entities.hashtags (array) : [string]
|-- entities.media (array)
|-- entities.media.additional_media_info (struct)
|-- entities.media.additional_media_info.description : string
|-- entities.media.additional_media_info.embeddable : boolean
|-- entities.media.additional_media_info.monetizable : bollean
|-- entities.media.diplay_url : string
|-- entities.media.id : long
|-- entities.media.id_str : string
|-- entities.urls (array)     
|-- extended_entities (struct)
|-- retweeted_status (struct)
|-- user (struct)

База данных Перейдите к: Data-178 KB. Затем скопируйте пронумерованные элементы в текстовый файл с именем "example". Сохраните в каталог с именем "../example.json/", созданный в вашем рабочем каталоге.

Код R записывается для воспроизведения примера, как показано ниже,

Выход из кода

library(sparklyr)
library(dplyr)
library(devtools)
  devtools::install_github("mitre/sparklyr.nested")
# If Spark is not installed, then also need:
# spark_install(version = "2.2.0")
library(sparklyr.nested)
library(testthat)
library(jsonlite)

Sys.setenv(SPARK_HOME="/usr/lib/spark")    
conf <- spark_config()
conf$'sparklyr.shell.executor-memory' <- "20g"
conf$'sparklyr.shell.driver-memory' <- "20g"
conf$spark.executor.cores <- 16
conf$spark.executor.memory <- "20G"
conf$spark.yarn.am.cores  <- 16
conf$spark.yarn.am.memory <- "20G"
conf$spark.executor.instances <- 8
conf$spark.dynamicAllocation.enabled <- "false"
conf$maximizeResourceAllocation <- "true"
conf$spark.default.parallelism <- 32
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) 
 sdf_schema_viewer(sample_tbl) # to create db schema

Усилия, предпринятые

Используется jsonlite. Но он не может читать большой файл и внутри кусков. Это заняло не окончание времени. Итак, я повернулся к Спарклыру, поскольку он задумывался и читал 1 миллиард записей за несколько секунд. Я сделал дальнейшее исследование для выравнивания записей до уровня глубокого гнездования (поскольку сплющивание выполняется в jsonlite пакете с помощью функции flatten flatten()). Но в Sparklyr такой возможности нет. В Спарклыре можно было только сгладить 1-й уровень.

Я хочу сгладить данные разных типов данных и хочу, чтобы результат в CSV файле.

Ответ 1

Хорошо, так что это один из возможных способов разогнать все это.

Вы можете использовать информацию о схеме для создания всех вложенных имен. Например, entities.media.additional_media_info, тогда вы можете просто использовать SQL для их выбора.

Это немного трудоемкий и не может обобщать, но он работает

Я хотел бы думать, что это тоже должно быть быстрым, поскольку это только SELECT.

columns_to_flatten <- sdf_schema_json(sample_tbl, simplify = T) %>%
  # using rlist package for ease of use
  rlist::list.flatten(use.names = T) %>%
  # get names
  names() %>%
  # remove contents of brackets and whitespace
  gsub("\\(.*?\\)|\\s", "", .) %>%
  # add alias for column names, dot replaced with double underscore
  # this avoids duplicate names that would otherwise occur with singular
  {paste(., "AS", gsub("\\.", "__", .))} %>%
  # required, otherwise doesn't seem to work
  sub("variants", "variants[0]", .)

# construct query
sql_statement <- paste("SELECT",
                       paste(columns_to_flatten, collapse = ", "),
                       "FROM example")

# execute on spark cluster, save as table in cluster
spark_session(sc) %>%
  sparklyr::invoke("sql", sql_statement) %>%
  sparklyr::invoke("createOrReplaceTempView", "flattened_example")

tbl(sc, "flattened_example") %>%
  sdf_schema_viewer()

Созданный SQL выглядит довольно просто, просто длинно:

SELECT contributors AS contributors, coordinates AS coordinates, created_at AS created_at, display_text_range AS display_text_range, entities.hashtags.indices AS entities__hashtags__indices, entities.hashtags.text AS entities__hashtags__text, entities.media.additional_media_info.description AS entities__media__additional_media_info__description, entities.media.additional_media_info.embeddable AS entities__media__additional_media_info__embeddable, entities.media.additional_media_info.monetizable AS entities__media__additional_media_info__monetizable, entities.media.additional_media_info.title AS entities__media__additional_media_info__title, entities.media.display_url AS entities__media__display_url, entities.media.expanded_url AS entities__media__expanded_url, entities.media.id AS entities__media__id, entities.media.id_str AS entities__media__id_str, entities.media.indices AS entities__media__indices, entities.media.media_url AS entities__media__media_url, entities.media.media_url_https AS entities__media__media_url_https, entities.media.sizes.large.h AS entities__media__sizes__large__h, entities.media.sizes.large.resize AS entities__media__sizes__large__resize, entities.media.sizes.large.w AS entities__media__sizes__large__w FROM example