Как можно добавить новый столбец в существующий Dataframe в Spark SQL

Question

Я использую DataFrame API.

У меня есть DataFrame и объект List (также можно использовать Array). Как можно добавить этот список в существующий DataFrame в качестве нового столбца? Должен ли я использовать класс Column для этого?

Answer 1

К сожалению, это была моя вина, я уже нашел функцию withColumn(String colName, Column col), которая должна решить мою проблему

Answer 2

Вы, вероятно, следует преобразовать список в один столбец РДУ и применять присоединиться на critetia pickeg вами. Простое преобразование DataFrame:

val df1 = sparkContext.makeRDD(yourList).toDF("newColumn") 

Если вам нужно создать дополнительный столбец для выполнения джойна на него можно добавлять столбцы, отображение списка:

val df1 = sparkContext.makeRDD(yourList).map(i => (i, fun(i)).toDF("newColumn", "joinOnThisColumn") 

Я не знаком с Java версии, но вы должны попробуйте использовать JavaSparkContext.parallelize(yourList) и примените аналогичные операции отображения на основе this doc.

Answer 3

Этот поток немного старый, но я столкнулся с аналогичной ситуацией с использованием Java. Я думаю больше всего на свете, было концептуальное непонимание того, как я должен подходить к этой проблеме.

Чтобы исправить мою проблему, я создал простую POJO, чтобы помочь с новым столбцом для набора данных (в отличие от попытки построить на существующем). Я думаю, что концептуально я не понял, что лучше всего создавать набор данных во время первоначального чтения, где нужно добавить дополнительный столбец. Надеюсь, это поможет кому-то в будущем.

Рассмотрим следующий пример:

 JavaRDD<MyPojo> myRdd = dao.getSession().read().jdbc("jdbcurl","mytable",someObject.getProperties()).javaRDD().map(new Function<Row,MyPojo>() { 

         private static final long serialVersionUID = 1L; 

         @Override 
         public MyPojo call(Row row) throws Exception { 
         Integer curDos = calculateStuff(row); //manipulate my data 

         MyPojo pojoInst = new MyPojo(); 

         pojoInst.setBaseValue(row.getAs("BASE_VALUE_COLUMN")); 
         pojoInst.setKey(row.getAs("KEY_COLUMN")); 
         pojoInst.setCalculatedValue(curDos); 

         return dos; 
         } 
        }); 

     Dataset<Row> myRddRFF = dao.getSession().createDataFrame(myRdd, MyPojo.class); 

//continue load or other operation here... 

Answer 4

Вот пример, где мы имели дату колонки и хотел бы добавить еще один столбец с месяц.

Dataset<Row> newData = data.withColumn("month", month((unix_timestamp(col("date"), "MM/dd/yyyy")).cast("timestamp"))); 

Надеюсь, это поможет!

Cheers!