Introdução ao pyspark

Exemplo de contagem de palavras no Pyspark

O exemplo subjacente é apenas o fornecido na documentação oficial do pyspark. Clique aqui para acessar este exemplo.

# the first step involves reading the source text file from HDFS 
text_file = sc.textFile("hdfs://...")

# this step involves the actual computation for reading the number of words in the file
# flatmap, map and reduceByKey are all spark RDD functions
counts = text_file.flatMap(lambda line: line.split(" ")) \
             .map(lambda word: (word, 1)) \
             .reduceByKey(lambda a, b: a + b)

# the final step is just saving the result.
counts.saveAsTextFile("hdfs://...")

Consumindo dados do S3 usando o PySpark

Existem dois métodos com os quais você pode consumir dados do bucket do AWS S3.

  1. Usando a API sc.textFile (ou sc.wholeTextFiles): Esta API também pode ser usada para HDFS e sistema de arquivos local.
aws_config = {}  # set your aws credential here
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", aws_config['aws.secret.access.key'])
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", aws_config['aws.secret.access.key'])
s3_keys = ['s3n/{bucket}/{key1}', 's3n/{bucket}/{key2}']
data_rdd = sc.wholeTextFiles(s3_keys)
  1. Lendo-o usando a API personalizada (digamos, um downloader de boto):
def download_data_from_custom_api(key):
    # implement this function as per your understanding (if you're new, use [boto][1] api)
    # don't worry about multi-threading as each worker will have single thread executing your job
    return ''

s3_keys = ['s3n/{bucket}/{key1}', 's3n/{bucket}/{key2}']
# numSlices is the number of partitions. You'll have to set it according to your cluster configuration and performance requirement
key_rdd = sc.parallelize(s3_keys, numSlices=16) 

data_rdd = key_rdd.map(lambda key: (key, download_data_from_custom_api(key))

Eu recomendo usar a abordagem 2 porque, ao trabalhar com a abordagem 1, o driver baixa todos os dados e os trabalhadores apenas os processam. Isso tem as seguintes desvantagens:

  1. Você ficará sem memória à medida que o tamanho dos dados aumentar.
  2. Seus funcionários ficarão ociosos até que os dados sejam baixados

Instalação ou Configuração

Instruções detalhadas sobre como configurar ou instalar o pyspark.