如何在Spark SQL中创建永久表

Question

如何在Spark SQL中创建永久表

4

在我的项目中，我正在将数据从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，sns数据约1K。我将这些表连接起来，但是这个执行时间非常高且不可接受。

- trallallalloo

2个回答

2

Spark SQL不是数据库，其中发生的数据操作只在创建它们的spark context可用时存在。有几个spark作业服务器实现可以使您保留一个作业的结果并发送其他作业以针对相同的数据集。它仍然是短暂的，如果服务器（即spark context）关闭，则必须重新加载。

话虽如此，您可以持久化计算结果并稍后检索它（将其返回到Mongo中，或者存储在Hadoop/其他文件系统中的文件中）。

- Arnon Rotem-Gal-Oz

谢谢您的回答。我明白Spark SQL表的生命周期与Spark Context的生命周期有限。那么，让我再问一个问题：我能减少查询执行时间吗？我正在独立模式下工作，以后我们将使用集群。当我们使用集群时，查询执行时间会减少到可接受的执行时间吗？ - trallallalloo

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Kaushal · Accepted Answer

是的，您可以通过缓存您的表、数据框或 Rdd 来提高程序执行时间。
如果您想将数据保存为永久表，则可以使用 df.saveAsTable 方法，但数据框应该是通过 HiveContext 创建的。
对于 JDBC 连接，您需要启动您的 Thrift 服务，然后可以对注册表执行 Spark Sql。