java - 如何在 Java 中并行运行 spark 程序
问题描述
所以我有一个具有 spark maven 依赖项的 java 应用程序,在运行它时,它会在运行它的主机上启动 spark 服务器。服务器实例有 36 个核心。我正在指定 SparkSession 实例,其中我提到了并行的核心数量和其他配置属性,但是当我看到使用 的统计信息时htop
,它似乎并没有使用所有核心,而只使用了 1 个。
SparkSession spark = SparkSession
.builder()
.master("local")
.appName("my-spark")
.config("spark.driver.memory","50g")
.config("spark.hadoop.fs.s3a.impl","org.apache.hadoop.fs.s3a.S3AFileSystem")
.config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")
.config("spark.sql.shuffle.partitions", "400")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "/dir1/dir2/logs")
.config("spark.history.fs.logDirectory", "/dir1/dir2/logs")
.config("spark.executor.cores", "36")
我还添加了 JavaSparkContext :
JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
sc.hadoopConfiguration().set("fs.s3a.access.key", AWS_KEY);
sc.hadoopConfiguration().set("fs.s3a.secret.key", AWS_SECRET_KEY);
sc.hadoopConfiguration().set("spark.driver.memory","50g");
sc.hadoopConfiguration().set("spark.eventLog.enabled", "true");
sc.hadoopConfiguration().set("spark.eventLog.dir", "/dir1/dir2/logs");
sc.hadoopConfiguration().set("spark.executor.cores", "36");
我的任务是将 aws s3 中的数据读取到 df 中并将数据写入另一个存储桶中。
Dataset<Row> df = spark.read().format("csv").option("header", "true").load("s3a://bucket/file.csv.gz");
//df = df.repartition(200);
df.withColumn("col_name", df.col("col_name")).sort("col_name", "_id").write().format("iceberg").mode("append").save(location);
解决方案
.gz 文件是“不可吐出的”:要解压缩它们,您必须从字节 0 开始并向前读取。因此,spark、hive、MapReduce 等将整个文件提供给单个工作人员。如果您想要并行处理,请使用不同的压缩格式(例如 snappy)
推荐阅读
- java - 解析文本文件并有效地拆分它
- javascript - jquery点击输入字段显示div,点击body隐藏div
- javascript - MS 图形 API:带有“/me/onlineMeetings”请求的 400 AuthenticationError
- firebase - 配置为单页应用(将所有网址重写为/index.html)后出现“出现意外错误”?是的
- maven - 将功能 Websphere Liberty 配置文件添加到应用程序
- node.js - 在重定向之前在 Express 应用中设置自定义引用者
- python - Google Calendar API 删除“conferenceData”嵌套对象
- arrays - 倒排数组(从最低索引读取)
- mysql - PRIMARY KEY 的所有部分都不能为空;如果您需要在键中使用 NULL,请改用 UNIQUE
- python - 如何修改列表中的多个项目?