PySpark é uma linguagem que permite aos usuários interagir com um back-end Apache Spark para processar dados rapidamente.

O Spark pode operar em grandes conjuntos de dados em uma rede distribuída de servidores, oferecendo grandes benefícios de desempenho e confiabilidade quando utilizado corretamente.

Este guia apresenta situações comuns que encontramos e as práticas recomendadas associadas.

Prefira a seleção de coluna implícita para acesso direto, exceto para desambiguação

# ruim
df = df.select(F.lower(df1.colA). F.upper(df2.colB))
# bom
df = df.select(F.lower(F.col('colA')), F.upper(F.col('colB')))
# melhor - desde Spark 3.0
df = df.select(F.lower('colA'), F.upper('colB'))

Na maioria das situações, é melhor evitar o primeiro e segundo estilos e apenas referenciar a coluna pelo nome, usando uma string, como no terceiro exemplo. O Spark 3.0 expandiu bastante os casos em que o terceiro estilo funciona. Quando o método string não é possível, entretanto, devemos recorrer a uma abordagem mais detalhada.

• Se o nome da variável do dataframe for grande, as expressões que o envolvem rapidamente se tornarão complicadas;

• Se o nome da coluna tiver um espaço ou outro caractere não suportado, o operador colchete deverá ser usado. Isso gera inconsistência, e df1['colA'] é tão difícil de escrever quanto F.col('colA');

• As expressões de coluna envolvendo o dataframe não são reutilizáveis e não podem ser usadas para definir funções abstratas;

• Renomear uma variável de dataframe pode ser propenso a erros, pois todas as referências de coluna devem ser atualizadas em conjunto.

Além disso, a sintaxe de ponto (".") incentiva o uso de nomes de variáveis curtos e não descritivos para os dfs.

Por outro lado, F.col('colA') sempre fará referência a uma coluna designada no dataframe que está sendo operado, denominado df neste caso.

Refatorar operações lógicas complexas

As operações lógicas, que geralmente residem dentro de .filter() ou F.when(), precisam ser legíveis. Aplicamos a mesma regra das funções de encadeamento, mantendo as expressões lógicas dentro do mesmo bloco de código em três (3) expressões no máximo. Se as operações crescem demais, muitas vezes é um sinal de que o código pode ser simplificado ou extraído. Extrair operações lógicas complexas em variáveis torna o código mais fácil de ler e raciocinar, o que também reduz bugs.

# ruim
F.when( (F.col('prod_status') == 'Delivered') | (((F.datediff('deliveryDate_actual', 'current_date') < 0) & ((F.col('currentRegistration') != '') | ((F.datediff('deliveryDate_actual', 'current_date') < 0) & ((F.col('originalOperator') != '') | (F.col('currentOperator') != '')))))), 'In Service')

O código acima pode ser simplificado de diversas maneiras. Para começar, concentre-se em agrupar as etapas lógicas em algumas variáveis nomeadas. O Pyspark requer que as expressões sejam colocadas entre parênteses. Isso, misturado com os parênteses utilizados para agrupar as operações lógicas pode prejudicar a legibilidade. O código acima tem uma redundância de difícil detecção dado a ilegibilidade do código :

(F.datediff(df.deliveryDate_actual, df.current_date) < 0)

# melhor
has_operator = ((F.col('originalOperator') != '') | (F.col('currentOperator') != ''))
delivery_date_passed = (F.datediff('deliveryDate_actual', 'current_date') < 0)
has_registration = (F.col('currentRegistration').rlike('.+'))
is_delivered = (F.col('prod_status') == 'Delivered')

F.when(is_delivered | (delivery_date_passed & (has_registration | has_operator)), 'In Service')

O exemplo acima elimina a expressão redundante e é mais fácil de ler, podemos melhorá-lo ainda mais reduzindo o número de operações.

# bom
has_operator = ((F.col('originalOperator') != '') | (F.col('currentOperator') != ''))
delivery_date_passed = (F.datediff('deliveryDate_actual', 'current_date') < 0)
has_registration = (F.col('currentRegistration').rlike('.+'))
is_delivered = (F.col('prod_status') == 'Delivered')
is_active = (has_registration | has_operator)

F.when(is_delivered | (delivery_date_passed & is_active), 'In Service')

Observe como a expressão F.when agora é sucinta e legível e o comportamento desejado é claro para qualquer pessoa que revise este código.

Instruções Select

Realizar o select() no começo de uma transformação PySpark é considerado uma boa prática, expressões envolvendo mais de um df, ou operações condicionais como .when() são desencorajadas de serem usadas em um select(), a menos que seja necessário por motivos de desempenho.

# ruim
aircraft = aircraft.select(
    'aircraft_id',
    'aircraft_msn',
    F.col('aircraft_registration').alias('registration'),
    'aircraft_type',
    F.avg('staleness').alias('avg_staleness'),
    F.col('number_of_economy_seats').cast('long'),
    F.avg('flight_hours').alias('avg_flight_hours'),
    'operator_code',
    F.col('number_of_business_seats').cast('long'),
)

A menos que a ordem seja importante para você, tente agrupar operações do mesmo tipo.

# bom
aircraft = aircraft.select(
    'aircraft_id',
    'aircraft_msn',
    'aircraft_type',
    'operator_code',
    F.col('aircraft_registration').alias('registration'),
    F.col('number_of_economy_seats').cast('long'),
    F.col('number_of_business_seats').cast('long'),
    F.avg('staleness').alias('avg_staleness'),
    F.avg('flight_hours').alias('avg_flight_hours'),
)

A instrução select() redefine o esquema de um dataframe, portanto, naturalmente suporta a inclusão ou exclusão de colunas, antigas e novas, bem como a redefinição de colunas pré-existentes. Ao centralizar todas essas operações em uma única instrução, fica muito mais fácil identificar o esquema final, o que torna o código mais legível e conciso.

Em vez de utilizar withColumnRenamed(), use aliases:

# ruim
df.select('key', 'comments').withColumnRenamed('comments', 'num_comments')

# bom
df.select('key', F.col('comments').alias('num_comments'))

Em vez de utilizar withColumn() para redefinir o tipo, converta no select:

# ruim
df.select('comments').withColumn('comments', F.col('comments').cast('double'))

# bom
df.select(F.col('comments').cast('double'))

Colunas Vazias