Apache Spark Java - как перебирать набор данных строки и удалять пустые поля - PullRequest
Новая
       29

Apache Spark Java - как перебирать набор данных строки и удалять пустые поля

0 голосов
/ 18 мая 2018

Я пытаюсь создать приложение spark, которое считывает данные из таблицы Hive, и вывод будет записан в формате JSON.

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

Я ожидаю, что мой вывод, как, пожалуйста, предложите, как я могу достичь этого? 

{"personId":"101","personName":"Sam","email":"Sam@gmail.com"}
{"personId":"102","personName":"Smith"}  // as email is null or blank should not be included in output

Вот мой код: 

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import com.fdc.model.Person;

public class ExtractionExample {

    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder().appName("ExtractionExample")
                .config("spark.sql.warehouse.dir", "/user/hive/warehouse/").enableHiveSupport().getOrCreate();
        Dataset<Row> sqlDF = spark.sql("SELECT person_id as personId, person_name as personName, email_id as emailId FROM person");
        Dataset<Person> person = sqlDF.as(Encoders.bean(Person.class));

        /*  
         * iterate through all the columns and identify the null value and drop
         * Looks like it will drop the column from entire table but when I tried it doesn't do anything.
         * String[] columns = sqlDF.columns();
        for (String column : columns) {
            String colValue = sqlDF.select(column).toString();
            System.out.println("printing the column: "+ column +" colvalue:"+colValue.toString());
            if(colValue != null && colValue.isEmpty() && (colValue).trim().length() == 0) {
                System.out.println("dropping the null value");
                sqlDF = sqlDF.drop(column);
            }

        }
        sqlDF.write().json("/data/testdb/test/person_json");
        */

        /* 
         * 
         * Unable to get the bottom of the solution 
         * also collect() is heavy operation is there any better way to do this?
         * List<Row> rowListDf = person.javaRDD().map(new Function<Row, Row>() {
                @Override
                public Row call(Row record) throws Exception {
                   String[] fieldNames =  record.schema().fieldNames();
                    Row modifiedRecord = new RowFactory().create();
                   for(int i=0; i < fieldNames.length; i++ ) {
                       String value = record.getAs(i).toString();
                      if (value!= null && !value.isEmpty() && value.trim().length() > 0) {
                          //   RowFactory.create(record.get(i)); ---> throwing this error
                      }
                   }
                    // return RowFactory object
                    return null;
                }
            }).collect();*/


        person.write().json("/data/testdb/test/person_json");

    }
}

1 Ответ

0 голосов
/ 18 мая 2018

Как и предполагалось, user9613318, средство записи JSON по умолчанию игнорирует поля NULL.

...