Funções de ordem superior, AVRO e serializadores personalizados

sparklyr 1.3 agora está disponível em Crancom os seguintes novos recursos novos:

• Funções de ordem superior para manipular facilmente matrizes e estruturas
• Suporte para Apache Avrouma estrutura de serialização de dados orientada a linha
• Serialização personalizada Usando as funções R para ler e escrever qualquer formato de dados
• Outras melhorias como compatibilidade com EMR 6.0 & Spark 3.0 e suporte inicial para a biblioteca de séries temporais de Flint

Para instalar sparklyr 1.3 de Cran, execute

Neste put up, destacaremos alguns novos recursos importantes introduzidos no Sparklyr 1.3 e nos cenários de exibição em que esses recursos são úteis. Enquanto uma série de aprimoramentos e correções de bugs (especialmente aquelas relacionadas a spark_apply()Assim, Arrow Apachee conexões de faísca secundárias) também foram uma parte importante deste lançamento, elas não serão o tópico deste put up, e será um exercício fácil para o leitor descobrir mais sobre eles do Sparklyr NOTÍCIAS arquivo.

Funções de ordem superior

Funções de ordem superior são construções Spark SQL incorporadas que permitem que as expressões lambda definidas pelo usuário sejam aplicadas com eficiência a tipos de dados complexos, como matrizes e estruturas. Como uma demonstração rápida para ver por que as funções de ordem superior são úteis, digamos um dia Scrooge McDuck mergulhou em seu enorme abóbado e encontrou grandes quantidades de centavos, níquel, moedas de dez centavos e quartos. Tendo um gosto impecável em estruturas de dados, ele decidiu armazenar as quantidades e valores de enfrentar tudo em duas colunas de matriz Spark SQL:

library(sparklyr)

sc <- spark_connect(grasp = "native", model = "2.4.5")
coins_tbl <- copy_to(
  sc,
  tibble::tibble(
    portions = checklist(c(4000, 3000, 2000, 1000)),
    values = checklist(c(1, 5, 10, 25))
  )
)

Declarando assim seu patrimônio líquido de centavos de 4K, níquel 3K, 2K dez centavos e 1k quartos. Para ajudar Scrooge McDuck a calcular o valor complete de cada tipo de moeda em Sparklyr 1.3 ou superior, podemos aplicar hof_zip_with()o equivalente sparklyr de Zip_withpara portions coluna e values coluna, combinando pares de elementos de matrizes nas duas colunas. Como você deve ter adivinhado, também precisamos especificar como combinar esses elementos e que melhor maneira de conseguir isso do que uma fórmula unilateral concisa~ .x * .yEm R, o que diz que queremos (quantidade * valor) para cada tipo de moeda? Então, temos o seguinte:

result_tbl <- coins_tbl %>%
  hof_zip_with(~ .x * .y, dest_col = total_values) %>%
  dplyr::choose(total_values)

result_tbl %>% dplyr::pull(total_values)

(1)  4000 15000 20000 25000

Com o resultado 4000 15000 20000 25000 Dizendo -nos que existem em US $ 40 dólares em um centavo, US $ 150 dólares em níquel, dólares de US $ 200 em dez centavos e US $ 250 dólares em trimestres, como esperado.

Usando outra função sparklyr nomeada hof_aggregate()que executa um AGREGAR Operação em Spark, podemos então calcular o patrimônio líquido de Scrooge McDuck com base em result_tblarmazenando o resultado em uma nova coluna chamada complete. Aviso para que esta operação agregada funcione, precisamos garantir que o valor inicial da agregação tenha tipo de dados (a saber, BIGINT) isso é consistente com o tipo de dados de total_values (que é ARRAY), como mostrado abaixo:

result_tbl %>%
  dplyr::mutate(zero = dplyr::sql("CAST (0 AS BIGINT)")) %>%
  hof_aggregate(begin = zero, ~ .x + .y, expr = total_values, dest_col = complete) %>%
  dplyr::choose(complete) %>%
  dplyr::pull(complete)

(1) 64000

Portanto, o patrimônio líquido de Scrooge McDuck é de US $ 640.

Outras funções de ordem superior suportadas pelo Spark SQL até agora incluem reworkAssim, filtere existsconforme documentado em aquie semelhante ao exemplo acima, seus colegas (ou seja, hof_transform()Assim, hof_filter()e hof_exists()) todos existem em sparklyr 1.3, para que possam ser integrados a outros dplyr verbos de maneira idiomática em R.

Avro

Outro destaque da versão Sparklyr 1.3 é o suporte interno para as fontes de dados AVRO. O Apache Avro é um protocolo de serialização de dados amplamente usado que combina a eficiência de um formato de dados binários com a flexibilidade das definições de esquema JSON. Para simplificar o trabalho com fontes de dados AVRO, no brilho 1.3, assim que uma conexão de faísca é instanciada com spark_connect(..., bundle = "avro")Sparklyr descobrirá automaticamente qual versão de spark-avro pacote para usar com essa conexão, economizando muitas dores de cabeça em potencial para usuários de brilho Sparklyr tentando determinar a versão correta de spark-avro por si mesmos. Semelhante a como spark_read_csv() e spark_write_csv() estão em vigor para trabalhar com dados de CSV, spark_read_avro() e spark_write_avro() Os métodos foram implementados no Sparklyr 1.3 para facilitar a leitura e a gravação de arquivos AVRO por meio de uma conexão Spark com capacidade AVRO, conforme ilustrado no exemplo abaixo:

library(sparklyr)

# The `bundle = "avro"` possibility is barely supported in Spark 2.4 or larger
sc <- spark_connect(grasp = "native", model = "2.4.5", bundle = "avro")

sdf <- sdf_copy_to(
  sc,
  tibble::tibble(
    a = c(1, NaN, 3, 4, NaN),
    b = c(-2L, 0L, 1L, 3L, 2L),
    c = c("a", "b", "c", "", "d")
  )
)

# This instance Avro schema is a JSON string that basically says all columns
# ("a", "b", "c") of `sdf` are nullable.
avro_schema <- jsonlite::toJSON(checklist(
  kind = "file",
  title = "topLevelRecord",
  fields = checklist(
    checklist(title = "a", kind = checklist("double", "null")),
    checklist(title = "b", kind = checklist("int", "null")),
    checklist(title = "c", kind = checklist("string", "null"))
  )
), auto_unbox = TRUE)

# persist the Spark knowledge body from above in Avro format
spark_write_avro(sdf, "/tmp/knowledge.avro", as.character(avro_schema))

# after which learn the identical knowledge body again
spark_read_avro(sc, "/tmp/knowledge.avro")

# Supply: spark (?? x 3)
      a     b c
    
  1     1    -2 "a"
  2   NaN     0 "b"
  3     3     1 "c"
  4     4     3 ""
  5   NaN     2 "d"

Serialização personalizada

Além dos formatos de serialização de dados comumente usados, como CSV, JSON, Parquet e Avro, a partir de Sparklyr 1.3, os procedimentos de serialização e desserialização do quadro de dados personalizados implementados em r também podem ser executados em trabalhadores de faísca por meio do recém -implementado spark_read() e spark_write() Métodos. Podemos ver os dois em ação através de um exemplo rápido abaixo, onde saveRDS() é chamado de uma função de escritor definida pelo usuário para salvar todas as linhas em um quadro de dados Spark em 2 arquivos RDS no disco e readRDS() é chamado de uma função de leitor definida pelo usuário para ler os dados dos arquivos RDS de volta a Spark:

library(sparklyr)

sc <- spark_connect(grasp = "native")
sdf <- sdf_len(sc, 7)
paths <- c("/tmp/file1.RDS", "/tmp/file2.RDS")

spark_write(sdf, author = operate(df, path) saveRDS(df, path), paths = paths)
spark_read(sc, paths, reader = operate(path) readRDS(path), columns = c(id = "integer"))

# Supply: spark> (?? x 1)
     id
  
1     1
2     2
3     3
4     4
5     5
6     6
7     7

Outras melhorias

Sparklyr.flint

Sparklyr.flint é uma extensão sparklyr que visa fazer funcionalidades do Pedra A biblioteca de séries temporais facilmente acessível a partir de R. está atualmente sob desenvolvimento ativo. Uma boa notícia é que, enquanto o authentic Pedra A biblioteca foi projetada para funcionar com Spark 2.x, um ligeiramente modificado garfo Dele funcionará bem com o Spark 3.0 e dentro da estrutura de extensão Sparklyr existente. sparklyr.flint Pode determinar automaticamente a qual versão da biblioteca Flint para carregar com base na versão do Spark está conectada. Outra boa notícia é, como mencionado anteriormente, sparklyr.flint Ainda não sabe muito sobre seu próprio destino. Talvez você possa desempenhar um papel ativo na formação de seu futuro!

EMR 6.0

Esta versão também apresenta uma mudança pequena, mas importante, que permite que o Sparklyr se conecte corretamente à versão do Spark 2.4, incluído no Amazon EMR 6.0.

Anteriormente, a Sparklyr assumiu automaticamente qualquer Spark 2.x a que estava se conectando, foi construído com o Scala 2.11 e tentou carregar os artefatos Scala necessários construídos com o Scala 2.11 também. Isso se tornou problemático ao se conectar ao Spark 2.4 da Amazon EMR 6.0, que é construído com Scala 2.12. A partir do Sparklyr 1.3, esse problema pode ser corrigido simplesmente especificando scala_version = "2.12" Ao ligar spark_connect() (por exemplo, spark_connect(grasp = "yarn-client", scala_version = "2.12")).

Spark 3.0

Por último, mas não menos importante, vale a pena mencionar Sparklyr 1.3.0 é conhecido por ser totalmente compatível com o Spark 3.0 lançado recentemente. É altamente recomendável atualizar sua cópia do Sparklyr para 1.3.0 se você planeja ter o Spark 3.0 como parte do seu fluxo de trabalho de dados no futuro.

Reconhecimento

Em ordem cronológica, queremos agradecer às seguintes indivíduos por enviar solicitações de tração para Sparklyr 1.3:

Também somos gratos por contribuições valiosas no roteiro Sparklyr 1.3, #2434e #2551 de (@javierluraschi)(https://github.com/javierluraschi), e um grande conselho espiritual sobre #1773 e #2514 de @mattpollock e @benmwhite.

Observe que, se você acredita que está faltando no reconhecimento acima, pode ser porque sua contribuição foi considerada parte da próxima liberação Sparklyr, em vez de parte da versão atual. Fazemos todos os esforços para garantir que todos os colaboradores sejam mencionados nesta seção. Caso você acredite que haja um erro, não hesite em entrar em contato com o autor desta postagem do weblog por e-mail (Yitao no RStudio Dot Com) e solicitar uma correção.

Se você deseja aprender mais sobre sparklyrrecomendamos visitar Sparklyr.aiAssim, Spark.rstudio.come alguns dos posts de lançamento anterior, como Sparklyr 1.2 e Sparklyr 1.1.

Obrigado pela leitura!