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

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

0 голосов
/ 06 сентября 2018

Введение

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)

База данных Перейдите к: Данные-178 КБ . Затем скопируйте пронумерованные элементы в текстовый файл с именем «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. Но он не может читать большие файлы, а также внутри. Это заняло бесконечное время. Итак, я повернулся к Sparklyr, поскольку он удивляется и прочитал 1 миллиард записей в течение нескольких секунд. Я проделал дальнейшее исследование по сглаживанию записей до уровня глубокой вложенности (поскольку сглаживание выполняется в пакете jsonlite с использованием функции flatten()). Но в Sparklyr такой функции нет. В Спарклире возможно только сплющивание 1-го уровня.

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

1 Ответ

0 голосов
/ 17 сентября 2018

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

Вы можете использовать информацию схемы для создания всех вложенных имен. Например, 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()

enter image description here

Сгенерированный 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
...