Я написал искровое задание (job-1), которое принимает входные данные из CSV-файлов, разделяет записи по идентификатору пользователя, а затем объединяет их в 100 сжатых сжатых файлов паркета. Сейчас я работаю над вторым заданием (задание-2), которое принимает выходные данные задания-1 и выполняет некоторые вычисления.
Ввод задания-2 имеет 2 столбца для широты и долготы (которые меня интересуют), а также идентификатор пользователя и метку времени. Я хочу создать два новых столбца, которые сместят широту и долготу на 1. Это потребуется для расчета пройденного расстояния по формуле Haversine . Я использую Java DataFrame API против Spark 2.4.0. Фрагмент кода выглядит примерно так:
Dataset<Row> ds1 = spark.read().parquet(path);
WindowSpec window = Window.partitionBy("idvalue").orderBy("timestamp");
ds1.na().drop(new String[] { "timestamp" });
Dataset<Row> ds2 = ds1.withColumn("hour", functions.hour(ds1.col("timestamp")))
.withColumn("day", functions.dayofmonth(ds1.col("timestamp")))
.withColumn("date", functions.to_date(ds1.col("timestamp")))
.withColumn("locationlat-shifted", functions.lag(ds1.col("locationlat"), 1).over(window))
.withColumn("locationlon-shifted", functions.lag(ds1.col("locationlon"), 1).over(window));
ds2.show(20);
Я делаю это правильно? Причина в том, что вывод информации занимает много времени (около 10 минут). Входные данные составляют около 2,5 ГБ. Я не уверен, что использую тот факт, что данные уже сгруппированы по идентификаторам пользователей во входных файлах паркета. В stdout я вижу эти два предупреждения:
WARN HttpMethodReleaseInputStream: Attempting to release HttpMethod in finalize() as its response data stream has gone out of scope. This attempt will not always succeed and cannot be relied upon! Please ensure response data streams are always fully consumed or closed to avoid HTTP connection starvation.
WARN HttpMethodReleaseInputStream: Successfully released HttpMethod in finalize(). You were lucky this time... Please ensure response data streams are always fully consumed or closed.
Я делаю раздел по идентификатору пользователя при создании windowspec, но не уверен, что это путь для продвижения вперед. Любое понимание приветствуется, поскольку я не очень хорошо знаком с операциями с окнами в spark-sql.