无法为大型数据集运行 Spark 作业

Posted 2023-04-18

【中文标题】无法为大型数据集运行 Spark 作业

【英文标题】：Cannot run Spark jobs for large datasets

【发布时间】：2017-06-12 03:41:38

【问题描述】：

我编写了一个 Spark 作业来读取 S3 中的 Hive 数据并生成 HFile。

这项工作在只读取一个 ORC 文件（大约 190 MB）时工作正常，但是，当我用它来读取整个 S3 目录时，大约 400 个 ORC 文件，所以大约 400*190 MB = 76 GB 数据，它保持抛出以下错误/堆栈跟踪：

17/06/12 01:48:03 ERROR server.TransportRequestHandler: Error sending result StreamResponsestreamId=/jars/importer-all.jar, byteCount=194727686, body=FileSegmentManagedBufferfile=/tmp/importer-all.jar, offset=0, length=194727686 to /10.211.XX.XX:39149; closing connection
java.nio.channels.ClosedChannelException
    at io.netty.channel.AbstractChannel$AbstractUnsafe.close(...)(Unknown Source)
17/06/12 01:48:03 WARN scheduler.TaskSetManager: Lost task 6.0 in stage 0.0 (TID 6, ip-10-211-127-63.ap-northeast-2.compute.internal, executor 9): java.nio.channels.ClosedChannelException
    at org.apache.spark.network.client.StreamInterceptor.channelInactive(StreamInterceptor.java:60)
    at org.apache.spark.network.util.TransportFrameDecoder.channelInactive(TransportFrameDecoder.java:179)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:251)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:237)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:230)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1289)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:251)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:237)
    at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:893)
    at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:691)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:408)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:455)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
    at java.lang.Thread.run(Thread.java:745)

17/06/12 01:48:03 INFO scheduler.TaskSetManager: Starting task 6.1 in stage 0.0 (TID 541, ip-10-211-126-250.ap-northeast-2.compute.internal, executor 72, partition 6, PROCESS_LOCAL, 6680 bytes)
17/06/12 01:48:03 ERROR server.TransportRequestHandler: Error sending result StreamResponsestreamId=/jars/importer-all.jar, byteCount=194727686, body=FileSegmentManagedBufferfile=/tmp/importer-all.jar, offset=0, length=194727686 to /10.211.XX.XX:39151; closing connection
java.nio.channels.ClosedChannelException
    at io.netty.channel.AbstractChannel$AbstractUnsafe.close(...)(Unknown Source)

我的集群足够大，可以处理它：（这已经验证过）

它有 40 个节点，超过 800 GB 的可用内存，320 个 VCore。

这是我的 Java 代码：

protected void sparkGenerateHFiles(JavaRDD<Row> rdd) throws IOException 
        System.out.println("In sparkGenerateHFiles....");
        JavaPairRDD<ImmutableBytesWritable, KeyValue> javaPairRDD = rdd.mapToPair(
            new PairFunction<Row, ImmutableBytesWritable, KeyValue>() 
            public Tuple2<ImmutableBytesWritable, KeyValue> call(Row row) throws Exception 
                System.out.println("running call now ....");
                String key = (String) row.get(0);
                String value = (String) row.get(1);

                ImmutableBytesWritable rowKey = new ImmutableBytesWritable();
                byte[] rowKeyBytes = Bytes.toBytes(key);
                rowKey.set(rowKeyBytes);

                KeyValue keyValue = new KeyValue(rowKeyBytes,
                    Bytes.toBytes("fam"),
                    Bytes.toBytes("qualifier"),
                    ProductJoin.newBuilder()
                        .setId(key)
                        .setSolrJson(value)
                        .build().toByteArray());

                return new Tuple2<ImmutableBytesWritable, KeyValue>(rowKey, keyValue);
            
        );
        Partitioner partitioner = new IntPartitioner(2);
        // repartition and sort the data - HFiles want sorted data
        JavaPairRDD<ImmutableBytesWritable, KeyValue> repartitionedRDD =
            javaPairRDD.repartitionAndSortWithinPartitions(partitioner);


        Configuration baseConf = HBaseConfiguration.create();
        Configuration conf = new Configuration();
        conf.set(HBASE_ZOOKEEPER_QUORUM, importerParams.zkQuorum);
        Job job = new Job(baseConf, "map data");
        HTable table = new HTable(conf, importerParams.hbaseTargetTable);
        System.out.println("gpt table: " + table.getName());
        HFileOutputFormat2.configureIncrementalLoad(job, table);
        System.out.println("Done configuring incremental load....");

        Configuration config = job.getConfiguration();


        repartitionedRDD.saveAsNewAPIHadoopFile(
            "HFILE_OUTPUT_PATH",
            ImmutableBytesWritable.class,
            KeyValue.class,
            HFileOutputFormat2.class,
            config
            );
        System.out.println("Saved to HFILE_OUTPUT_PATH: " + HFILE_OUTPUT_PATH);
    

protected JavaRDD<Row> readJsonTable() 
        System.out.println("In readJsonTable.....");
        SparkSession.Builder builder = SparkSession.builder().appName("Importer");
        String hiveTable = "";
        if (importerParams.local) 
            builder.master("local");
            hiveTable = HIVE_TABLE_S3A_DEV_SAMPLE;
         else 
            hiveTable = importerParams.hiveSourceTable;
        
        SparkSession spark = builder.getOrCreate();

        SparkContext sparkContext = spark.sparkContext();

        // this is important. need to set the endpoint to ap-northeast-2
        sparkContext.hadoopConfiguration()
            .set("fs.s3a.endpoint", "s3.ap-northeast-2.amazonaws.com");

        Dataset<Row> rows = null;
        if (importerParams.local) 
            rows = spark.read().format("orc").load(hiveTable);
         else 
            rows = spark.read().format("orc").load(hiveTable);//use this one temporarily
//          rows = spark.read().format("orc").load(HIVE_TABLE_S3A_PREFIX
            // + importerParams.latestDateHour);
        
        System.out.println("Finished loading hive table from S3, rows.count() = "
            + (rows != null ? rows.count() : 0));

        return rows.toJavaRDD();
    

主程序：

        long startTime = System.currentTimeMillis();
        JavaRDD<Row> rdd = readJsonTable();

        sparkGenerateHFiles(rdd);
        System.out.println("it took " + (System.currentTimeMillis() - startTime)/1000 + " seconds to generate HFiles...\n\n\n\n");

我尝试过的：

我在 *** 上看到了最接近的 post。 然后我设置了这个 builder.config("spark.shuffle.blockTransferService", "nio"); 但仍然没有运气。

非常感谢任何帮助！

【问题讨论】：

可能是数据倾斜导致了异常。

【参考方案1】：

正如@Wang 指出的那样，这确实是由于我的数据倾斜问题。

为了解决这个问题，我做的是：

我重新创建了我的 HBase 表，但这次我使用了SPLITS，并将我的 HBase 表拆分为 80 个区域。 然后在我的 Spark 代码中，我编写了一个自定义的 Partitioner 来根据每个条目的键重新分区，这样就不会出现HOTSPOTTING 问题，即一个区域服务器过载而其他区域服务器空闲。

沿途学到的一些其他技巧，在使用SPLITS创建HBase表时，默认情况下第一个region的startkey和最后一个region的endkey是空字符串""，一定要在那里做正确的事情以避免热点。

这是我的partitioner 的一个工作示例。

谢谢！

【讨论】：

这是大数据中非常常见的问题，寻找好的键来避免长尾分布。

@FisherCoder 请更新自定义分区器示例的链接