Как создать постоянный стол в spark sql

Question

в моем проекте, я переношу данные из MongoDB в таблицу SparkSQL для запросов на основе SQL. Но Spark SQL позволяет мне создавать временные файлы. Когда я хочу что-то запросить, время выполнения очень велико, потому что операция передачи и отображения данных занимает слишком много времени.

Итак, можно ли сократить время выполнения? Могу ли я создавать постоянные таблицы Spark SQL? Можно ли запрашивать постоянные таблицы с JDBC?

Я добавляю свой код и результаты выполнения. Я делаю все в автономном режиме.

package com.mongodb.spark.sql; 

import java.util.List; 

import org.apache.hadoop.conf.Configuration; 
import org.apache.spark.SparkConf; 
import org.apache.spark.api.java.JavaRDD; 
import org.apache.spark.api.java.JavaSparkContext; 
import org.apache.spark.api.java.function.Function; 
import org.apache.spark.sql.DataFrame; 
import org.apache.spark.sql.Row; 
import org.apache.spark.sql.SQLContext; 
import org.bson.BSONObject; 

import com.mongodb.hadoop.MongoInputFormat; 
import com.mongodb.spark.demo.Observation; 
import com.mongodb.spark.demo.Sensor; 

import scala.Tuple2; 

public class SparkSqlMongo { 

public static void main(String[] args) { 

    Configuration conf = new Configuration(); 

    conf.set("mongo.job.input.format", "com.mongodb.hadoop.MongoInputFormat"); 
    conf.set("mongo.input.uri", "mongodb://localhost:27017/test.observations"); 

    Configuration sensConf = new Configuration(); 

    sensConf.set("mongo.job.input.format", "com.mongodb.hadoop.MongoInputFormat"); 
    sensConf.set("mongo.input.uri", "mongodb://localhost:27017/test.sens"); 

    SparkConf sconf = new SparkConf().setMaster("local[2]").setAppName("SQL DENEME").set("nsmc.connection.host", 
      "mongodb:"); 

    JavaSparkContext sc = new JavaSparkContext(sconf); 
    SQLContext sql = new SQLContext(sc); 

    JavaRDD<Observation> obs = sc.newAPIHadoopRDD(conf, MongoInputFormat.class, Object.class, BSONObject.class) 
      .map(new Function<Tuple2<Object, BSONObject>, Observation>() { 

       private static final long serialVersionUID = 1L; 

       @Override 
       public Observation call(Tuple2<Object, BSONObject> v1) throws Exception { 

        int id = (int) v1._2.get("_id"); 
        double value = (double) v1._2.get("Value"); 
        // Date time = (Date) v1._2.get("Time"); 
        int sensor = (int) v1._2.get("SensorId"); 
        int stream = (int) v1._2.get("DataStreamId"); 

        Observation obs = new Observation(id, value, sensor, stream); 
        return obs; 

       } 
      }); 

    DataFrame obsi = sql.createDataFrame(obs, Observation.class); 

    obsi.registerTempTable("obsi"); 

    JavaRDD<Sensor> sens = sc.newAPIHadoopRDD(sensConf, MongoInputFormat.class, Object.class, BSONObject.class) 
      .map(new Function<Tuple2<Object, BSONObject>, Sensor>() { 

       /** 
       * 
       */ 
       private static final long serialVersionUID = 1L; 

       @Override 
       public Sensor call(Tuple2<Object, BSONObject> v1) throws Exception { 

        int id = (int) v1._2.get("_id"); 
        String name = (String) v1._2.get("Name"); 
        String description = (String) v1._2.get("Description"); 

        Sensor s = new Sensor(id, name, description); 

        System.out.println(s.getName()); 
        return s; 

       } 
      }); 

    DataFrame sensi = sql.createDataFrame(sens, Sensor.class); 

    sensi.registerTempTable("sensi"); 

    sensi.show(); 

    long start = System.currentTimeMillis(); 

    DataFrame obser = sql 
      .sql("SELECT obsi.value, obsi.id, sensi.name FROM obsi, sensi WHERE obsi.sensorID = sensi.id and sensi.id = 107") 
      .cache(); 
    long stop = System.currentTimeMillis(); 

    // System.out.println("count ====>>> " + a.toString()); 
    System.out.println("toplam sorgu zamani : " + (stop - start)); 
    ; 
    // 
    // while(!obser.equals(null)){ 
    // System.out.println(obser); 
    // } 

    List<String> names = obser.javaRDD().map(new Function<Row, String>() { 

     private static final long serialVersionUID = 1L; 

     public String call(Row row) { 

      // System.out.println(row); 
      // System.out.println("value : " + row.getDouble(0) + " id : " + 
      // row.getInt(1) + " name : " + row.getString(0)); 
      return "Name: " + row; 
     } 
    }).collect(); 

} 

}

Все время выполнения составляет около 120 секунд в течение около 5M наблюдения и 1К SNS данных. Я присоединяюсь к этим таблицам, и это время исполнения очень велико и неприемлемо.

Answer 1

1. Да, вы можете улучшить время выполнения программы на Caching your Table, Dataframe или Rdd.
2. И, если вы хотите сохранить данные как постоянную таблицу, вы можете использовать метод df.saveAsTable, но dataframe должен быть создан через HiveContext.
3. Для подключения JDBC необходимо запустить Thrift service, после чего вы можете выполнить команду Spark Sql на таблицах регистров.

Answer 2

Spark SQL не манипулирует данными базы данных, которые происходят внутри него, только до тех пор, пока доступен контекст искры, который их создал. Существует несколько реализаций сервера искровых заданий, которые позволят вам удерживать результат одного задания и отправлять другие задания против одного и того же набора данных. Он будет по-прежнему временным и должен быть перезагружен, если сервер (т.е. контекст искры) отключается

Это говорит о том, что вы можете сохранить результат вычисления и получить его позже (либо обратно в Монго, либо в файлы на Hadoop/other файловая система)