我是Spark的新手，我正在EMR集群上运行spark作业。完成这项工作大约需要一个小时。
在job中，有一段代码用于从S3阅读文件（文件大小为10MB）并将其内容写入dynamo db。
我不明白为什么要花这么长时间才能完成这项任务？我以为它会更快地完成。
下面是Spark Job的代码：

public class PopulateCovid19Citations {

    private static final String APP_NAME = "PopulateCovid19Citations";
    private static Logger LOGGER = LogManager.getLogger(PopulateCovid19Citations.class);
    static Logger log = Logger.getLogger(PopulateCovid19Citations.class.getName());

    public static void main(String[] args) throws Exception {

        Logger.getLogger("org").setLevel(Level.ERROR);

        // Building the Spark session
        JavaSparkContext javaSparkContext = SparkConfiguration.buildSparkContext(APP_NAME);
        SparkSession sparkSession = SparkSession.builder()
            .sparkContext(javaSparkContext.sc()).getOrCreate();

        String fileLocationTest = "s3a://mybucket/WHOCovid19CitationsDatabase.csv";
        Dataset<Row> fullCitations =
            sparkSession.read().format("com.databricks.spark.csv").option("inferSchema", "true")
                .option("header", "true").load(fileLocationTest);
        fullCitations.show(10);

        // Selecting only the relevant columns for our exercise
        Dataset<Row> citations = fullCitations.select(
            col("Title"),
            col("Authors"),
            col("Abstract"),
            col("Published Year"),
            col("Journal"),
            col("Study"),
            col("Tags"));
        citations.show(10);

        // Removing citations with null title
        Dataset<Row> filteredCitations = citations.filter(col("Title").isNotNull());

        // Building a RDD composed of DynamoDB writable items that matches the Covid19Citation table
        JavaPairRDD<Text, DynamoDBItemWritable> dynamoCitations = filteredCitations.javaRDD().mapToPair(citation -> {
            Map<String, AttributeValue> attributes = new HashMap<>();
            putStringAttribute(attributes, "citationCode", UUID.randomUUID().toString());
            putStringAttribute(attributes, "title", citation.getAs("Title"));
            putStringAttribute(attributes, "authors", citation.getAs("Authors"));
            putStringAttribute(attributes, "abstract", citation.getAs("Abstract"));
            putNumberAttribute(attributes, "publishedYear", citation.getAs("Published Year"));
            putStringAttribute(attributes, "journal", citation.getAs("Journal"));
            putStringAttribute(attributes, "study", citation.getAs("Study"));
            putStringAttribute(attributes, "tags", citation.getAs("Tags"));

            DynamoDBItemWritable dynamoDBItemWritable = new DynamoDBItemWritable();
            dynamoDBItemWritable.setItem(attributes);
            return new Tuple2<>(new Text(""), dynamoDBItemWritable);
        });

        // Writing data to the DynamoDB table
        JobConf jobConf = SparkConfiguration.buildJobConf(javaSparkContext, Boolean.FALSE);
        dynamoCitations.saveAsHadoopDataset(jobConf);
        sparkSession.stop();
    }

    private static void putStringAttribute(Map<String, AttributeValue> attributes, String key, Object fieldValue) {
        if (fieldValue != null) {
            attributes.put(key, new AttributeValue(fieldValue.toString()));
        }
    }

    private static void putNumberAttribute(Map<String, AttributeValue> attributes, String key, Object fieldValue) {
        if (fieldValue != null) {
            try {
                Integer integerFieldValue = Integer.parseInt(fieldValue.toString());
                AttributeValue attributeValue = new AttributeValue();
                attributeValue.setN(integerFieldValue.toString());
                attributes.put(key, attributeValue);
            } catch (Exception e) {
                LOGGER.info("cannot convert " + fieldValue + " to integer");
            }
        }
    }
}

下面是构建Spark上下文的配置：

public static JavaSparkContext buildSparkContext(String application) throws ClassNotFoundException {
        SparkConf conf = new SparkConf()
                .setAppName(application)
                  .set("spark.executor.cores", "5")
                  .set("spark.executor.memory", "42G")
                  .set("spark.executor.memoryOverhead", "4G")
                  .set("spark.driver.memory", "42G")
                  .set("spark.driver.cores", "5")
                  .set("spark.executor.instances", "25")
                  .set("spark.default.parallelism", "250")
                  .set("spark.dynamicAllocation.enabled", "false")
                .registerKryoClasses(new Class<?>[]{
                        Class.forName("org.apache.hadoop.io.Text"),
                        Class.forName("org.apache.hadoop.dynamodb.DynamoDBItemWritable")
                });
        return new JavaSparkContext(conf);
    }

我正在使用带有m6a.16xlarge（64vcore，259GB内存）的EMR集群用于EC2示例类型。
有人能帮我一下吗？
先谢谢你了。