Uma extensão de série temporal para sparklyr

Nesta postagem do weblog, mostraremos sparklyr.flintum novíssimo sparklyr extensão que fornece uma interface R simples e intuitiva para o Flint biblioteca de séries temporais. sparklyr.flint está disponível em CRAN hoje e pode ser instalado da seguinte forma:

set up.packages("sparklyr.flint")

As duas primeiras seções desta postagem serão uma rápida visão panorâmica sobre sparklyr e Flinto que garantirá aos leitores não familiarizados com sparklyr ou Flint podemos ver ambos como blocos de construção essenciais para sparklyr.flint. Depois disso, apresentaremos sparklyr.flintfilosofia de design, estado atual, exemplos de uso e, por último, mas não menos importante, suas direções futuras como um projeto de código aberto nas seções subsequentes.

sparklyr é uma interface R de código aberto que integra o poder da computação distribuída de Apache Faísca com os idiomas, ferramentas e paradigmas familiares para transformação e modelagem de dados em R. Ele permite que pipelines de dados que funcionam bem com dados não distribuídos em R sejam facilmente transformados em análogos que podem processar dados distribuídos em grande escala no Apache Spark .

Em vez de resumir tudo sparklyr tem a oferecer em poucas frases, o que é impossível de fazer, esta seção se concentrará apenas em um pequeno subconjunto de sparklyr funcionalidades que são relevantes para conectar-se ao Apache Spark a partir de R, importar dados de séries temporais de fontes de dados externas para o Spark e também transformações simples que normalmente fazem parte das etapas de pré-processamento de dados.

Conectando-se a um cluster Apache Spark

O primeiro passo para usar sparklyr é conectar-se ao Apache Spark. Geralmente isso significa um dos seguintes:

• Executando o Apache Spark localmente em sua máquina e conectando-se a ele para testar, depurar ou executar demonstrações rápidas que não exigem um cluster Spark de vários nós:

• Conectando-se a um cluster Apache Spark de vários nós gerenciado por um gerenciador de cluster como FIOpor exemplo,

  library(sparklyr)
  
  sc <- spark_connect(grasp = "yarn-client", spark_home = "/usr/lib/spark")

Importando dados externos para Spark

Disponibilizar dados externos no Spark é fácil com sparklyr dado o grande número de fontes de dados sparklyr suporta. Por exemplo, dado um dataframe R, como

o comando para copiá-lo para um dataframe Spark com 3 partições é simplesmente

sdf <- copy_to(sc, dat, identify = "unique_name_of_my_spark_dataframe", repartition = 3L)

Da mesma forma, também existem opções para ingestão de dados em CSV, JSON, ORC, AVRO e muitos outros formatos conhecidos no Spark:

sdf_csv <- spark_read_csv(sc, identify = "another_spark_dataframe", path = "file:///tmp/file.csv", repartition = 3L)
  # or
  sdf_json <- spark_read_json(sc, identify = "yet_another_one", path = "file:///tmp/file.json", repartition = 3L)
  # or spark_read_orc, spark_read_avro, and so forth

Transformando um dataframe Spark

Com sparklyra maneira mais simples e legível de transformar um dataframe Spark é usando dplyr verbos e o operador pipe (%>%) de Magrittr.

Sparklyr suporta um grande número de dplyr verbos. Por exemplo,

Garante sdf contém apenas linhas com IDs não nulos e, em seguida, eleva ao quadrado o worth coluna de cada linha.

Isso é tudo para uma rápida introdução ao sparklyr. Você pode aprender mais em sparklyr.aionde você encontrará hyperlinks para materiais de referência, livros, comunidades, patrocinadores e muito mais.

Flint é uma poderosa biblioteca de código aberto para trabalhar com dados de série temporal no Apache Spark. Em primeiro lugar, ele suporta o cálculo eficiente de estatísticas agregadas em pontos de dados de séries temporais com o mesmo carimbo de knowledge/hora (também conhecido como summarizeCycles em Flint nomenclatura), dentro de uma determinada janela de tempo (também conhecida como, summarizeWindows), ou dentro de alguns intervalos de tempo (também conhecidos como summarizeIntervals). Ele também pode unir dois ou mais conjuntos de dados de série temporal com base na correspondência inexata de carimbos de knowledge/hora usando funções de junção asof, como LeftJoin e FutureLeftJoin. O autor de Flint descreveu muitos mais Flintprincipais funcionalidades do este artigoque achei extremamente útil ao descobrir como construir sparklyr.flint como uma interface R simples e direta para tais funcionalidades.

Os leitores que desejam alguma experiência prática direta com Flint e Apache Spark podem seguir as etapas a seguir para executar um exemplo mínimo de uso do Flint para analisar dados de série temporal:

• Primeiro, instale o Apache Spark localmente e, por motivos de conveniência, defina o SPARK_HOME variável de ambiente. Neste exemplo, executaremos o Flint com o Apache Spark 2.4.4 instalado em ~/sparkentão:

  export SPARK_HOME=~/spark/spark-2.4.4-bin-hadoop2.7

• Inicie o shell Spark e instrua-o a fazer o obtain Flint e suas dependências do Maven:

  "${SPARK_HOME}"/bin/spark-shell --packages=com.twosigma:flint:0.6.0

• Crie um dataframe simples do Spark contendo alguns dados de série temporal:

  import spark.implicits._
  
  val ts_sdf = Seq((1L, 1), (2L, 4), (3L, 9), (4L, 16)).toDF("time", "worth")

• Importe o dataframe junto com metadados adicionais, como unidade de tempo e nome da coluna de carimbo de knowledge/hora, para um TimeSeriesRDDpara que Flint pode interpretar os dados da série temporal de forma inequívoca:

  import com.twosigma.flint.timeseries.TimeSeriesRDD
  
  val ts_rdd = TimeSeriesRDD.fromDF(
    ts_sdf
  )(
    isSorted = true, // rows are already sorted by time
    timeUnit = java.util.concurrent.TimeUnit.SECONDS,
    timeColumn = "time"
  )

• Finalmente, depois de todo o trabalho árduo acima, podemos aproveitar várias funcionalidades de série temporal fornecidas por Flint analisar ts_rdd. Por exemplo, o seguinte produzirá uma nova coluna chamada value_sum. Para cada linha, value_sum conterá a soma de worths que ocorreram nos últimos 2 segundos a partir do carimbo de knowledge/hora dessa linha:

  import com.twosigma.flint.timeseries.Home windows
  import com.twosigma.flint.timeseries.Summarizers
  
  val window = Home windows.pastAbsoluteTime("2s")
  val summarizer = Summarizers.sum("worth")
  val consequence = ts_rdd.summarizeWindows(window, summarizer)
  
  consequence.toDF.present()

    +-------------------+-----+---------+
    |               time|worth|value_sum|
    +-------------------+-----+---------+
    |1970-01-01 00:00:01|    1|      1.0|
    |1970-01-01 00:00:02|    4|      5.0|
    |1970-01-01 00:00:03|    9|     14.0|
    |1970-01-01 00:00:04|   16|     29.0|
    +-------------------+-----+---------+

Em outras palavras, dado um carimbo de knowledge/hora t e uma linha no resultado tendo time igual a tpode-se notar o value_sum coluna dessa linha contém a soma de worthestá dentro da janela de tempo de (t - 2, t) de ts_rdd.

O propósito de sparklyr.flint é fazer funcionalidades de série temporal de Flint facilmente acessível a partir sparklyr. Para ver sparklyr.flint em ação, pode-se folhear o exemplo da seção anterior, seguir o seguinte para produzir o R-equivalente exato de cada etapa desse exemplo e, em seguida, obter o mesmo resumo do resultado ultimate:

• Primeiro de tudo, instale sparklyr e sparklyr.flint se você ainda não fez isso.

• Conecte-se ao Apache Spark que está sendo executado localmente em sparklyrmas lembre-se de anexar sparklyr.flint antes de correr sparklyr::spark_connecte, em seguida, importe nossos dados de série temporal de exemplo para o Spark:

• Converter sdf acima em um TimeSeriesRDD

  ts_rdd <- fromSDF(sdf, is_sorted = TRUE, time_unit = "SECONDS", time_column = "time")

• E finalmente, execute o resumidor ‘soma’ para obter um somatório de worths em todas as janelas de tempo dos últimos 2 segundos:

  consequence <- summarize_sum(ts_rdd, column = "worth", window = in_past("2s"))
  
  print(consequence %>% gather())

  ## # A tibble: 4 x 3
  ##   time                worth value_sum
  ##                      
  ## 1 1970-01-01 00:00:01     1         1
  ## 2 1970-01-01 00:00:02     4         5
  ## 3 1970-01-01 00:00:03     9        14
  ## 4 1970-01-01 00:00:04    16        29

A alternativa para fazer sparklyr.flint um sparklyr extensão é agrupar todas as funcionalidades de série temporal que ela fornece sparklyr em si. Decidimos que isso não seria uma boa ideia pelos seguintes motivos:

• Nem todos sparklyr os usuários precisarão dessas funcionalidades de série temporal
• com.twosigma:flint:0.6.0 e todos os pacotes Maven dos quais ele depende transitivamente são bastante pesados ​​em termos de dependência
• Implementando uma interface R intuitiva para Flint também pega um número não trivial de arquivos de origem R e torna tudo isso parte do sparklyr em si seria demais

Então, considerando tudo o que foi dito acima, construir sparklyr.flint como uma extensão de sparklyr parece ser uma escolha muito mais razoável.

Recentemente sparklyr.flint teve seu primeiro lançamento de sucesso no CRAN. No momento, sparklyr.flint só suporta o summarizeCycle e summarizeWindow funcionalidades de Flinte ainda não oferece suporte a asof be a part of e outras operações úteis de série temporal. Enquanto sparklyr.flint contém interfaces R para a maioria dos resumidores em Flint (pode-se encontrar a lista de resumidores atualmente suportados por sparklyr.flint em aqui), ainda faltam alguns deles (por exemplo, o suporte para OLSRegressionSummarizerentre outros).

Em geral, o objetivo de construir sparklyr.flint é que seja uma fina “camada de tradução” entre sparklyr e Flint. Deve ser o mais simples e intuitivo possível, ao mesmo tempo que suporta um rico conjunto de Flint funcionalidades de série temporal.

Acolhemos cordialmente qualquer contribuição de código aberto para sparklyr.flint. Por favor visite https://github.com/r-spark/sparklyr.flint/points se você gostaria de iniciar discussões, relatar bugs ou propor novos recursos relacionados a sparklyr.flinte https://github.com/r-spark/sparklyr.flint/pulls se você gostaria de enviar solicitações pull.

• Em primeiro lugar, o autor deseja agradecer a Javier (@javierluraschi) por propor a ideia de criar sparklyr.flint como a interface R para Flinte por sua orientação sobre como construí-lo como uma extensão do sparklyr.

• Tanto Javier (@javierluraschi) e Daniel (@dfalbel) ofereceram inúmeras dicas úteis sobre como fazer o envio inicial de sparklyr.flint para CRAN com sucesso.

• Nós realmente apreciamos o entusiasmo de sparklyr usuários que estavam dispostos a dar sparklyr.flint uma tentativa emblem após seu lançamento no CRAN (e houve alguns downloads de sparklyr.flint na semana passada, de acordo com as estatísticas do CRAN, o que foi bastante encorajador para nós). Esperamos que você goste de usar sparklyr.flint.

• O autor também agradece as valiosas sugestões editoriais de Mara (@batpigandme), Sigrid (@skeydan) e Javier (@javierluraschi) nesta postagem do weblog.

Obrigado por ler!