Объединенный вывод Spark в один файл

Question

Мне интересно, есть ли способ объединить конечный результат в один файл при использовании Spark? Вот код, который у меня есть:

conf = SparkConf().setAppName("logs").setMaster("local[*]")
sc = SparkContext(conf = conf)

logs_1 = sc.textFile('logs/logs_1.tsv')
logs_2 = sc.textFile('logs/logs_2.tsv')

url_1 = logs_1.map(lambda line: line.split("\t")[2])
url_2 = logs_2.map(lambda line: line.split("\t")[2])

all_urls = uls_1.intersection(urls_2)
all_urls = all_urls.filter(lambda url: url != "localhost") 

all_urls.collect()

all_urls.saveAsTextFile('logs.csv')

Кажется, что метод collect () не работает (или я неправильно понял его назначение). По сути, мне нужен «saveAsTextFile» для вывода в один файл, а не в папку с частями.

Answer 1

Ниже приведены некоторые предложения:

• collect() и saveAsTextFile() - это действия, которые означают, что они будут собирать результаты на узле драйвера.Поэтому излишне называть их обоих.

• В вашем случае вам просто нужно сохранить данные с saveAsTextFile(), вам не нужно звонить collect().

• collect() возвращает массив элементов (в вашем случае вы не используете возвращаемую переменную)

• Как только что предложили Гленни и Акашиспользуйте coalesce(1), чтобы форсировать один раздел.coalesce(1) не вызовет тасования, следовательно, гораздо более эффективно.

• В данном коде вы используете RDD API Spark, я бы предложил вместо этого использовать наборы данных / наборы данных.

Пожалуйста, обратитесь к следующим ссылкам для получения дополнительной информации о RDD и фреймах данных:

Разница между DataFrame, Dataset и RDD в Spark

https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html

Answer 2

Ну, прежде чем сохранить, вы можете один раз перераспределить, как показано ниже:

all_urls.repartition(1).saveAsTextFile(resultPath)

тогда вы получите только один файл результатов.

Answer 3

Вы можете сохранить его в формате parquet.Это лучший формат, подходящий для HDFS

all_urls.write.parquet("dir_name")