1

我正在做简单的 spark 聚合操作，从 avro 文件中读取数据作为数据帧，然后使用 rdd.map 方法将它们映射到案例类，然后进行一些聚合操作，比如 count 等。大多数时候它工作得很好。但有时它会产生奇怪的 CodeGen 异常；

[ERROR] 2017-03-24 08:43:20,595 org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator logError - failed to compile: java.lang.NullPointerException
/* 001 */ public java.lang.Object generate(Object[] references) {
/* 002 */   return new SpecificUnsafeProjection(references);
/* 003 */ }
/* 004 */
/* 005 */ class SpecificUnsafeProjection extends org.apache.spark.sql.catalyst.expressions.UnsafeProjection {

我正在使用此代码；

    val deliveries = sqlContext.read.format("com.databricks.spark.avro").load(deliveryDir)
                      .selectExpr("FROM_UNIXTIME(timestamp/1000, 'yyyyMMdd') as day",
                        "FROM_UNIXTIME(timestamp/1000, 'yyyyMMdd_HH') as hour",
                        "deliveryId"
                       )
                      .filter("valid = true").rdd
                      .map(row => {
                        val deliveryId = row.getAs[Long]("deliveryId")
                        val uid = row.getAs[Long]("uid")
                        val deviceModelId: Integer = if(row.getAs[Integer]("deviceModelId") == null) {
                          0
                        } else {
                          row.getAs[Integer]("deviceModelId")
                        }
                        val delivery = new DeliveryEvent(deliveryId, row.getAs[Integer]("adId"), row.getAs[Integer]("adSpaceId"), uid, deviceModelId)
                        eventCache.getDeliverCache().put(new Element(deliveryId, delivery))
                        new InteractedAdInfo(row.getAs[String]("day"), delivery.deliveryId, delivery.adId, delivery.adSpaceId, uid, deviceModelId, deliveryEvent=1)
                      })
deliveries.count()

我无法重现问题。但我在生产中不定期地得到它。我正在使用 java-app 并获取 spark-core_2.11:2.1.0 和 spark-avro_2.11:3.1.0 maven 坐标。

问题可能出在哪里，我在运行应用程序时正在设置 java -Xms8G -Xmx12G -XX:PermSize=1G -XX:MaxPermSize=1G 。

4

1 回答 1

0

我在非常简单的操作中看到了类似的错误，spark.read.format("com.databricks.spark.avro").load(fn).cache.count当应用于大型 AVRO 文件（我的测试中为 4GB-10GB 范围）时，这是间歇性的。但是，我可以消除删除设置--conf spark.executor.cores=4并将其默认为 1 的错误。 

WARN TaskSetManager: Lost task 58.0 in stage 2.0 (TID 82, foo.com executor 10): java.lang.RuntimeException: 
Error while encoding: 
java.util.concurrent.ExecutionException: 
java.lang.Exception: failed to compile: java.lang.NullPointerException
/* 001 */ public java.lang.Object generate(Object[] references) {
/* 002 */   return new SpecificUnsafeProjection(references);
/* 003 */ }
于 2017-11-01T19:41:42.260 回答