Não sei se foi só comigo, mas...
No arquivo de sócios part-00005 tem um registro de data inválido que é o "4100813", precisei dar um jeito de tratar esse valor como nulo na conversão de datas. Estava dando alguns erros comigo então já aproveitei e tratei todos os outros dataframes do spark assim também pra evitar erros! :)
Oi, Vagner! Como vai?
Agradeço por compartilhar sua observação com a comunidade Alura.
Tratar valores inválidos como nulos é uma ótima prática e ajuda bastante na robustez do pipeline. Excelente iniciativa em aplicar o mesmo tratamento para todos os DataFrames. Parabéns pela postura proativa!
Qualquer dúvida que surgir, compartilhe no fórum. Abraços e bons estudos!
Conte com o apoio da comunidade Alura na sua jornada. Abraços e bons estudos!
Olá Vagner! Estou enfrentando este mesmo problema. De que maneira você está tratando esta data inválida?
Olá Diogo!
Para resolver, primeiro eu deixei tudo como texto ao carregar os dados no dataframe em seguida utilizei mais ou menos a seguinte ideia para todas colunas de data de todos os dataframes, depois disso não tive mais problemas
from pyspark.sql import functions as f
from pyspark.sql.types import DateType
from datetime import datetime
def converter_data(data_str):
try:
return datetime.strptime(str(data_str), '%Y%m%d').date()
except:
return None
converter_udf = f.udf(converter_data, DateType())
df = df.withColumn("data_coluna", converter_udf("data_coluna"))
deve ter algum outro jeito melhor, mas esse resolveu meu problema :D
Obrigado por compartilhar, Wagner.
Um pouco antes da sua resposta, consegui da seguinte forma:
socios = socios.withColumn('data_de_entrada_sociedade', f.try_to_timestamp(socios.data_de_entrada_sociedade.cast(StringType()), f.lit('yyyyMMdd')))
Desta forma o campo fica com o tipo timestamp. Ele seta nulo para os inválidos. Depois é só converter novamente para DateType.
Por nada!
Resolvi fazer um teste no código que me mandou e funciona também perfeitamente! (Irá retornar um timestamp o que também serve)
De qualquer forma quem precisar usar tanto sua função quanto:
def converter_data(data_str):
try:
return datetime.strptime(str(data_str), '%Y%m%d').date()
except:
return None
converter_udf = f.udf(converter_data, DateType())
## Converter Colunas Datas::
socios = socios.withColumn( 'data_de_entrada_sociedade', converter_udf(socios.data_de_entrada_sociedade) )
estabelecimentos = estabelecimentos\
.withColumn(
'data_situacao_cadastral', converter_udf(estabelecimentos.data_situacao_cadastral),
)\
.withColumn(
'data_de_inicio_atividade', converter_udf(estabelecimentos.data_de_inicio_atividade),
)\
.withColumn(
'data_da_situacao_especial', converter_udf(estabelecimentos.data_da_situacao_especial),
)
socios.show(5)
socios.printSchema()
Vai funcionar perfeitamente!
Só deixar a coluna de data como string antes ou ja trazer o df como string! :)