如何在 spark sql 中创建永久表

Posted 2023-03-27

【中文标题】如何在 spark sql 中创建永久表

【英文标题】：how to create permanent table in spark sql

【发布时间】：2015-07-31 06:23:19

【问题描述】：

在我的项目中，我将数据从 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();

对于大约 5M 的观察和 1K 的 sns 数据，所有执行时间约为 120 秒。我加入了这些表，这个执行时间非常长并且无法接受。

【参考方案1】：

是的，您可以通过Caching your Table、Dataframe 或 Rdd 来提高程序执行时间。 而且，如果您想将数据保存为永久表，则可以使用 df.saveAsTable 方法，但应通过 HiveContext 创建数据框。 对于 JDBC 连接，您需要启动 Thrift service，然后您可以在寄存器表上执行 Spark Sql。

【讨论】：

感谢您的回答。我研究了 Hive，查询性能并没有让我满意。所以，我正在寻找更有效的方法。最后，我使用 parquet 文件来存储数据。当我查询时，我从 parquet 查询中获取数据，这种方式将查询性能提高了大约 4-5 倍。

【参考方案2】：

Spark SQL 不是数据库数据操作，只要创建它们的 spark 上下文可用，就会在其中发生数据操作。有几种 Spark 作业服务器实现可以让您保存一个作业的结果并针对同一数据集发送其他作业。如果服务器（即火花上下文）关闭，它仍然是暂时的并且必须重新加载

也就是说，您可以保留计算结果并稍后检索（返回 Mongo，放入 Hadoop/其他文件系统上的文件中）

【讨论】：

感谢您的回答。我了解 Spark SQL 表的生命周期受 Spark Context 生命周期的限制。那么，让我再问一个问题：我可以减少查询执行时间吗？我正在使用独立模式，稍后我们将使用集群。当我们使用集群时，查询执行时间会减少到可以接受的执行时间吗？