Как сделать параллельную обработку в pyspark

Question

Я хочу выполнить параллельную обработку для l oop с использованием pyspark.

from pyspark.sql import SparkSession
spark = SparkSession.builder.master('yarn').appName('myAppName').getOrCreate()
spark.conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

data = [a,b,c]


for i in data:
    try:
        df = spark.read.parquet('gs://'+i+'-data')
        df.createOrReplaceTempView("people")
        df2=spark.sql("""select * from people """)
        df.show()
    except Exception as e:
        print(e)
        continue

Вышеупомянутый скрипт работает нормально, но я хочу выполнить параллельную обработку в pyspark и это возможно в scala

Answer 1

Spark it-self запускает работу параллельно, но если вам все еще нужно параллельное выполнение в коде, вы можете использовать простой код python для параллельной обработки, чтобы сделать это.

data = ["a","b","c"]

from multiprocessing.pool import ThreadPool
pool = ThreadPool(10)


def fun(x):
    try:
        df = sqlContext.createDataFrame([(1,2, x), (2,5, "b"), (5,6, "c"), (8,19, "d")], ("st","end", "ani"))
        df.show()
    except Exception as e:
        print(e)

pool.map( fun,data)

Я изменил ваш код a немного, но это в основном то, как вы можете запускать параллельные задачи. Если у вас есть несколько плоских файлов, которые вы хотите запустить параллельно, просто составьте список с их именем и передайте его в pool.map (fun, data) .

Измените функцию fun , если это необходимо.

Для получения подробной информации о режиме в многопроцессорном модуле см. Документацию .

Аналогично, если вы хотите сделать это в Scala, вам потребуются следующие модули

import scala.concurrent.{Future, Await}

Для более детального понимания проверьте this out. Код предназначен для Databricks, но с некоторыми изменениями он будет работать в вашей среде.