小编use*_*662的帖子

我想从hdfs位置读取一堆文本文件,并使用spark在迭代中对其执行映射.

JavaRDD<String> records = ctx.textFile(args[1], 1); 能够一次只读取一个文件.

我想读取多个文件并将它们作为单个RDD处理.怎么样？

我有一个类如下,当我通过命令行运行时,我想看到进度状态.就像是,

10% completed... 
30% completed... 
100% completed...Job done!

我在纱线上使用spark 1.0并使用Java API.

public class MyJavaWordCount {
    public static void main(String[] args) throws Exception {
        if (args.length < 2) {
            System.err.println("Usage: MyJavaWordCount <master> <file>");
            System.exit(1);
        }
        System.out.println("args[0]: <master>="+args[0]);
        System.out.println("args[1]: <file>="+args[1]);

        JavaSparkContext ctx = new JavaSparkContext(
                args[0],
                "MyJavaWordCount",
                System.getenv("SPARK_HOME"),
                System.getenv("SPARK_EXAMPLES_JAR"));
        JavaRDD<String> lines = ctx.textFile(args[1], 1);

//      output                                            input   output         
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            //              output       input 
            public Iterable<String> call(String s) {
                return Arrays.asList(s.split(" "));
            }
        });

//          K       V                                                input …

我的用例如下所述.

1. 使用sparkContext.textFile(输入路径)从本地文件系统读取输入数据.
2. 在将输入数据提交到mapper/reducer函数之前,使用RDD.coalesce(numberOfPArtitions)将输入数据(8000万条记录)分区为分区.如果不对输入数据使用coalesce()或repartition(),则执行的速度非常慢,并且会因内存不足而失败.

我在这里面临的问题是决定要在输入数据上应用的分区数.输入数据大小每次都在变化,硬编码特定值不是一种选择.只有当某些最佳分区应用于我必须执行大量迭代(试验和错误)的输入数据时,spark才能很好地执行.这不是生产环境中的选项.

我的问题:根据输入数据大小和可用的集群资源(执行程序,核心等等),是否有拇指规则来决定所需的分区数量？如果是,请指出我的方向.任何帮助深表感谢.

我在纱线上使用火花1.0.

谢谢,AG