Prezados leitores, bom dia!
Neste artigo, o consultor sênior da divisão de negócios Big Data Solutions da empresa "Neoflex" descreve detalhadamente as opções para construir vitrines de estrutura variável utilizando o Apache Spark.
No âmbito de projetos de análise de dados, muitas vezes surge a tarefa de construir vitrines a partir de dados pouco estruturados.
Normalmente, esses dados são logs ou respostas de diversos sistemas, armazenadas em formato JSON ou XML. Os dados são exportados para o Hadoop e, a partir deles, deve-se construir a vitrine. Podemos organizar o acesso à vitrine criada, por exemplo, através do Impala.
Nesse caso, o esquema da vitrine alvo é previamente desconhecido. Além disso, o esquema não pode ser elaborado com antecedência, pois depende dos dados, e lidamos exatamente com esses dados pouco estruturados.
Por exemplo, hoje foi registrado a seguinte resposta:
{source: "app1", error_code: ""}e amanhã, do mesmo sistema, chega a seguinte resposta:
{source: "app1", error_code: "error", description: "Erro de rede"}Como resultado, deve-se adicionar mais um campo à vitrine — descrição, e não se sabe se ele virá ou não.
A tarefa de criar vitrines com esses dados é bastante padrão, e o Spark tem uma série de ferramentas para isso. Para fazer o parsing dos dados de origem, há suporte tanto para JSON quanto para XML, e para um esquema desconhecido previamente, há suporte para schemaEvolution.
À primeira vista, a solução parece simples. Basta pegar uma pasta com JSON e ler em um dataframe. O Spark criará o esquema e converterá dados aninhados em estruturas. Em seguida, tudo deve ser salvo em parquet, que é suportado inclusive no Impala, registrando a vitrine no Hive metastore.
Parece tudo simples.
No entanto, a partir de exemplos curtos na documentação, não é claro como lidar com uma série de problemas na prática.
A documentação descreve uma abordagem não para a criação da vitrine, mas para a leitura de JSON ou XML em um dataframe.
Concretamente, apenas é mostrado como ler e parsear JSON:
df = spark.read.json(path...)Isso é suficiente para tornar os dados disponíveis para o Spark.
Na prática, o cenário é muito mais complicado do que simplesmente ler arquivos JSON de uma pasta e criar um dataframe. A situação se apresenta assim: já existe uma vitrine definida, novos dados chegam todos os dias, e é necessário adicioná-los à vitrine, sem esquecer que o esquema pode diferir.
O esquema habitual para a construção da vitrine é o seguinte:
Passo 1. Os dados são carregados no Hadoop com uma atualização diária subsequente e armazenados em uma nova partição. Assim, obtém-se uma pasta particionada por dias com os dados de origem.
Passo 2. Durante o carregamento inicial, essa pasta é lida e analisada pelos meios do Spark. O dataframe obtido é salvo em um formato acessível para análise, como parquet, que pode ser importado posteriormente no Impala. Isso cria uma vitrine de destino com todos os dados acumulados até aquele momento.
Passo 3. É criada uma atualização que vai atualizar a vitrine todos os dias.
Surge a questão do carregamento incremental, a necessidade de particionamento da vitrine, e a questão do suporte ao esquema geral da vitrine.
Vamos dar um exemplo. Suponha que o primeiro passo da construção do repositório tenha sido implementado e que a exportação de arquivos JSON para a pasta esteja configurada.
Criar um dataframe a partir deles para depois salvar como vitrine não é um problema. Este é o primeiro passo, que pode ser facilmente encontrado na documentação do Spark:
df = spark.read.option("mergeSchema", True).json(".../*")
df.printSchema()
root
|-- a: long (nullable = true)
|-- b: string (nullable = true)
|-- c: struct (nullable = true) |
|-- d: long (nullable = true)Parece que está tudo certo.
Nós lemos e analisamos o JSON; agora salvamos o dataframe como parquet, registrando no Hive de qualquer maneira conveniente:
df.write.format(“parquet”).option('path','').saveAsTable('') Obtemos a vitrine.
Porém, no dia seguinte, novos dados foram adicionados da fonte. Temos uma pasta com JSON, e uma vitrine criada com base nessa pasta. Após o carregamento do próximo conjunto de dados da fonte, a vitrine está faltando dados de um dia.
Uma solução lógica seria particionar a vitrine por dias, o que permite adicionar uma nova partição a cada dia subsequente. Isso também é um mecanismo bem conhecido, e o Spark permite gravar partições separadamente.
Primeiro, fazemos o carregamento inicial, salvando os dados como descrito acima, apenas adicionando o particionamento. Esta ação é chamada de inicialização da vitrine e é feita apenas uma vez:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
No dia seguinte, carregamos apenas a nova partição:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/" )
Resta apenas reenregistrar no Hive para atualizar o esquema.
No entanto, é aqui que surgem os problemas.
O primeiro problema. Tarde ou cedo, o parquet resultante não poderá ser lido. Isso se deve à forma como parquet e JSON lidam com campos vazios de maneira diferente.
Vamos considerar uma situação típica. Por exemplo, ontem recebemos o JSON:
Dia 1: {"a": {"b": 1}},
e hoje o mesmo JSON se apresenta assim:
Dia 2: {"a": null}
Suponha que temos duas partições diferentes, cada uma contendo uma linha.
Quando lemos os dados originais como um todo, o Spark consegue determinar o tipo e entende que "a" é um campo do tipo "estrutura", com um campo interno "b" do tipo INT. No entanto, se cada partição foi salva separadamente, teremos um parquet com esquemas de partições incompatíveis:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Essa situação é bem conhecida, por isso foi adicionada uma opção — ao fazer o parse dos dados originais, remover os campos vazios:
df = spark.read.json("...", dropFieldIfAllNull=True)
Nesse caso, o parquet será composto por partições que poderão ser lidas juntas.
Embora aqueles que já tenham feito isso na prática sorriam amargamente. Por quê? Porque provavelmente surgirão mais duas situações. Ou três. Ou quatro. A primeira, que surgirá quase com certeza, é que os tipos numéricos aparecerão de maneira diferente em diferentes arquivos JSON. Por exemplo, {intField: 1} e {intField: 1.1}. Se esses campos estiverem em uma mesma partição, a fusão dos esquemas irá ler tudo corretamente, convertendo para o tipo mais preciso. Mas, se estiverem em partições diferentes, em uma teremos intField: int, e na outra intField: double.
Para lidar com essa situação, há a seguinte flag:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Agora temos uma pasta onde estão as partições que podem ser lidas em um único dataframe e um parquet válido para toda a vitrine. Certo? Errado.
É preciso lembrar que registramos a tabela no Hive. O Hive não é sensível a maiúsculas e minúsculas nos nomes dos campos, mas o parquet é sensível. Portanto, as partições com esquemas: field1: int, e Field1: int são iguais para o Hive, mas não para o Spark. É necessário não esquecer de converter os nomes dos campos para minúsculas.
Após isso, parece que está tudo bem.
No entanto, não é tão simples. Surge um segundo problema, também bem conhecido. Como cada nova partição é salva separadamente, na pasta da partição estarão arquivos de sistema do Spark, como o arquivo de sucesso da operação _SUCCESS. Isso levará a um erro ao tentar usar o parquet. Para evitar isso, é necessário configurar a configuração para proibir o Spark de escrever os arquivos de sistema na pasta:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Parece que agora, todo dia, uma nova partição parquet está sendo adicionada à pasta da vitrine-alvo, onde os dados processados do dia são armazenados. Nos preocupamos previamente para que não haja partições com conflitos de tipos de dados.
Mas, agora, temos o terceiro problema. Agora, o esquema geral não é conhecido e, além disso, na tabela do Hive, o esquema está incorreto, pois cada nova partição provavelmente distorceu o esquema.
Precisamos reRegistrar a tabela. Isso pode ser feito de forma simples: ler novamente a vitrine parquet, pegar o esquema e criar, com base nisso, o DDL, com o qual registraremos novamente a pasta no Hive como uma tabela externa, atualizando o esquema da vitrine-alvo.
Diante disso, surge o quarto problema. Quando registramos a tabela pela primeira vez, baseamo-nos no Spark. Agora estamos fazendo isso sozinhos, e é importante lembrar que os campos parquet podem começar com caracteres que não são permitidos no Hive. Por exemplo, o Spark descarta linhas que não conseguiu processar no campo "corrupt_record". Este campo não poderá ser registrado no Hive sem a devida escape.
Sabendo disso, obtemos o esquema:
f_def = ""
for f in pf.dtypes:
if f[0] != "date_load":
f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<")
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '‹user›/admin/testJson/testSchemaEvolution/pq/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)
Código ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") faz um DDL seguro, ou seja, em vez de:
create table tname (_field1 string, 1field string)
Com nomes de campos como "_field1, 1field", é criado um DDL seguro, onde os nomes dos campos são escapados: create table `tname` (`_field1` string, `1field` string).
Surge a questão: como obter corretamente um dataframe com o esquema completo (no código pf)? Como obter esse pf? Este é o quinto problema. ReLer o esquema de todas as partições na pasta com os arquivos parquet da vitrine-alvo? Este método é o mais seguro, mas difícil.
O esquema já está no Hive. Para obter um novo esquema, você pode combinar o esquema de toda a tabela com a nova partição. Portanto, é necessário obter o esquema da tabela do Hive e combiná-lo com o esquema da nova partição. Isso pode ser feito lendo os metadados de teste do Hive, salvando-os em uma pasta temporária e lendo ambas as partições de uma só vez com o Spark.
Basicamente, temos tudo o que precisamos: o esquema original da tabela no Hive e a nova partição. Também temos os dados. Resta apenas obter o novo esquema, que combina o esquema da vitrine com os novos campos da partição criada:
from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")
Em seguida, criamos o DDL para registrar a tabela, como no fragmento anterior.
Se toda a cadeia funcionar corretamente, ou seja, se houve um carregamento inicial e a tabela estiver corretamente criada no Hive, então obtemos o esquema atualizado da tabela.
E, por último, o problema é que não é tão simples adicionar uma partição à tabela do Hive, pois ela ficará corrompida. É necessário fazer com que o Hive ajuste sua estrutura de partições:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Uma tarefa simples de leitura de JSON e criação de uma vitrine a partir dele resulta na superação de uma série de dificuldades implícitas, cujas soluções precisam ser buscadas separadamente. E embora essas soluções sejam simples, leva-se muito tempo para encontrá-las.
Para implementar a construção da vitrine, foi necessário:
- Adicionar partições à vitrine, livrando-se de arquivos de controle
- Resolver campos vazios nos dados de origem que o Spark tipificou
- Converter tipos simples para string
- Ajustar os nomes dos campos para letras minúsculas
- Separar a extração de dados do registro da tabela no Hive (criação do DDL)
- Lembrar de escapar os nomes dos campos que podem ser incompatíveis com o Hive
- Aprender a atualizar o registro da tabela no Hive
Em resumo, a solução para a construção de vitrines contém muitos desafios ocultos. Portanto, ao enfrentar dificuldades na implementação, é melhor consultar um parceiro experiente com um histórico de sucesso.
Agradecemos por ler este artigo e esperamos que as informações sejam úteis.
Fonte: habr.com
Compre hospedagem confiável para sites com proteção contra DDoS, servidores VPS VDS 🔥 Compre hospedagem confiável para sites com proteção contra DDoS, servidores VPS VDS | ProHoster
