Я определил функцию:
def setJsonPushIndex(spark: SparkSession, currentdate: String, jsonPushInfo: RDD[(String, String)]): RDD[String] =
{
val sc = spark.sparkContext
val acc = new LongAccumulator()
sc.register(acc, "myaccumulator")
val jsonPushWithIndex = jsonPushInfo.map(x =>
{
acc.add(1)
val sendhour = x._2.toString
val index = pushUtil.toIndex(acc.value.toString)
var mid = "BI" + currentdate + sendhour + index
if (sendhour.toInt < pushUtil.getNextHour().toInt)
{
mid = "BI" + pushUtil.getNextday() + sendhour + index
}
mid + "\u0001" +
"\"mid\": " + "\"" + mid + "\"," +
x._1
}
)
jsonPushWithIndex
}
, затем я вызываю основную функцию:
val json_push_res = setJsonPushIndex(spark, currentdate, json_pushInfo)
val jsonResultPath= "/jc/algorithm/NewUserPushTest/results/" + pushUtil.NowDate() + "/"
json_push_res.take(12).foreach(println)
json_push_res.saveAsTextFile(jsonResultPath)
оказывается, в моем каталоге только один раздел "part-00000",моя команда:
spark-submit --master yarn --num-executors 5 --executor-cores 2 --executor-memory 5G --driver-memory 10G --class "apppush.NewUserPush_V2" /home/ilambda/lyj/test2.jar
номер результата около 30000.
Является ли номер результата маленьким, так как наконец-то есть один раздел или Accumulator
приводит к этому результату?