Почему select после объединения вызывает исключение в java spark dataframe? - PullRequest
Новая
       1

Почему select после объединения вызывает исключение в java spark dataframe?

3 голосов
/ 19 марта 2019

У меня есть два кадра данных: левый и правый.Они одинаковы и состоят из трех столбцов: src relation, dest и имеют одинаковые значения.

1- Я попытался объединить эти кадры данных, где условие - это dst слева = src справа.Но это не сработало.Где ошибка? 

Dataset<Row> r = left
  .join(right, left.col("dst").equalTo(right.col("src")));

Результат: 

+---+---------+---+---+---------+---+
|src|predicate|dst|src|predicate|dst|
+---+---------+---+---+---------+---+
+---+---------+---+---+---------+---+

2 - Если я переименовал dst слева как dst, а столбец src справа как dst2, то яприменить соединение, это работает.Но если я попытаюсь выбрать какой-то столбец из открытого кадра данных.Это поднимает исключение.Где моя ошибка? 

 Dataset<Row> left = input_df.withColumnRenamed("dst", "dst2");
 Dataset<Row> right = input_df.withColumnRenamed("src", "dst2");  
 Dataset<Row> r = left.join(right, left.col("dst2").equalTo(right.col("dst2")));

Тогда: 

left.show();

дает: 

+---+---------+----+
|src|predicate|dst2|
+---+---------+----+
|  a|       r1| :b1|
|  a|       r2|   k|
|:b1|       r3| :b4|
|:b1|      r10|   d|
|:b4|       r4|   f|
|:b4|       r5| :b5|
|:b5|       r9|   t|
|:b5|      r10|   e|
+---+---------+----+

и 

right.show();

дает: 

+----+---------+---+
|dst2|predicate|dst|
+----+---------+---+
|   a|       r1|:b1|
|   a|       r2|  k|
| :b1|       r3|:b4|
| :b1|      r10|  d|
| :b4|       r4|  f|
| :b4|       r5|:b5|
| :b5|       r9|  t|
| :b5|      r10|  e|
+----+---------+---+

результат: 

+---+---------+----+----+---------+---+
|src|predicate|dst2|dst2|predicate|dst|
+---+---------+----+----+---------+---+
|  a|       r1| b1| b1  |      r10|  d|
|  a|       r1| b1| b1  |       r3| b4|
|b1 |       r3| b4| b4  |       r5| b5|
|b1 |       r3| b4| b4  |       r4|  f|
+---+---------+----+----+---------+---+


Dataset<Row> r = left
  .join(right, left.col("dst2").equalTo(right.col("dst2")))
  .select(left.col("src"),right.col("dst"));

результат: 

Exception in thread "main" org.apache.spark.sql.AnalysisException: resolved attribute(s) dst#45 missing from dst2#177,src#43,predicate#197,predicate#44,dst2#182,dst#198 in operator !Project [src#43, dst#45];

3 - предположим, выбранные работы, как можно добавить полученный кадр данных в левый кадр данных.

Я работаю на Java.

1 Ответ

1 голос
/ 19 марта 2019

Вы использовали: 

r = r.select(left.col("src"), right.col("dst"));

Похоже, что Spark не находит происхождение обратно к right кадру данных.Не шокирует, поскольку проходит много оптимизаций.

Предполагая, что желаемый результат: 

+---+---+
|src|dst|
+---+---+
| b1|:b5|
| b1|  f|
|:b4|  e|
|:b4|  t|
+---+---+

Вы можете использовать один из следующих 3 вариантов:

Используя метод col() 

Dataset<Row> resultOption1Df = r.select(left.col("src"), r.col("dst"));
resultOption1Df.show();

Использование col() статической функции 

Dataset<Row> resultOption2Df = r.select(col("src"), col("dst"));
resultOption2Df.show();

Использование имен столбцов 

Dataset<Row> resultOption3Df = r.select("src", "dst");
resultOption3Df.show();

Вот полный исходный код: 

package net.jgp.books.spark.ch12.lab990_others;

import static org.apache.spark.sql.functions.col;

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

/**
 * Self join.
 * 
 * @author jgp
 */
public class SelfJoinAndSelectApp {

  /**
   * main() is your entry point to the application.
   * 
   * @param args
   */
  public static void main(String[] args) {
    SelfJoinAndSelectApp app = new SelfJoinAndSelectApp();
    app.start();
  }

  /**
   * The processing code.
   */
  private void start() {
    // Creates a session on a local master
    SparkSession spark = SparkSession.builder()
        .appName("Self join")
        .master("local[*]")
        .getOrCreate();

    Dataset<Row> inputDf = createDataframe(spark);
    inputDf.show(false);

    Dataset<Row> left = inputDf.withColumnRenamed("dst", "dst2");
    left.show();

    Dataset<Row> right = inputDf.withColumnRenamed("src", "dst2");
    right.show();

    Dataset<Row> r = left.join(
        right,
        left.col("dst2").equalTo(right.col("dst2")));
    r.show();

    Dataset<Row> resultOption1Df = r.select(left.col("src"), r.col("dst"));
    resultOption1Df.show();

    Dataset<Row> resultOption2Df = r.select(col("src"), col("dst"));
    resultOption2Df.show();

    Dataset<Row> resultOption3Df = r.select("src", "dst");
    resultOption3Df.show();
  }

  private static Dataset<Row> createDataframe(SparkSession spark) {
    StructType schema = DataTypes.createStructType(new StructField[] {
        DataTypes.createStructField(
            "src",
            DataTypes.StringType,
            false),
        DataTypes.createStructField(
            "predicate",
            DataTypes.StringType,
            false),
        DataTypes.createStructField(
            "dst",
            DataTypes.StringType,
            false) });

    List<Row> rows = new ArrayList<>();
    rows.add(RowFactory.create("a", "r1", ":b1"));
    rows.add(RowFactory.create("a", "r2", "k"));
    rows.add(RowFactory.create("b1", "r3", ":b4"));
    rows.add(RowFactory.create("b1", "r10", "d"));
    rows.add(RowFactory.create(":b4", "r4", "f"));
    rows.add(RowFactory.create(":b4", "r5", ":b5"));
    rows.add(RowFactory.create(":b5", "r9", "t"));
    rows.add(RowFactory.create(":b5", "r10", "e"));

    return spark.createDataFrame(rows, schema);
  }
}
...