小编sim*_*yun的帖子

有没有人如何在spark sql查询中进行分页？

我需要使用spark sql但不知道如何进行分页.

尝试:

select * from person limit 10, 10

我有一个RMI集群.每个RMI服务器都有一个Spark上下文.有没有办法在不同的Spark上下文之间共享RDD？

我的代码算法如下
Step1.获取一个hbase实体数据到hBaseRDD

      JavaPairRDD<ImmutableBytesWritable, Result> hBaseRDD = 
                 jsc.newAPIHadoopRDD(hbase_conf,  TableInputFormat.class,
                 ImmutableBytesWritable.class, Result.class); 

第2步.将hBaseRDD转换为rowPairRDD

     // in the rowPairRDD the key is hbase's row key, The Row is the hbase's Row data 
     JavaPairRDD<String, Row> rowPairRDD = hBaseRDD 
                            .mapToPair(***); 
    dataRDD.repartition(500);
        dataRDD.cache();

Step3.将rowPairRDD转换为schemaRDD

            JavaSchemaRDD schemaRDD =   sqlContext.applySchema(rowPairRDD.values(), schema); 
            schemaRDD.registerTempTable("testentity"); 
           sqlContext.sqlContext().cacheTable("testentity");

Step4.使用spark sql做第一个简单的sql查询.

   JavaSQLContext  sqlContext = new org.apache.spark.sql.api.java.JavaSQLContext(jsc);
    JavaSchemaRDD retRDD=sqlContext.sql("SELECT column1, column2 FROM testentity WHERE 
             column3 = 'value1' ") 
     List<org.apache.spark.sql.api.java.Row> rows = retRDD.collect(); 

Step5.使用spark sql做第二个简单的sql查询.

JavaSchemaRDD retRDD=sqlContext.sql("SELECT column1, column2 FROM testentity 
                                     WHERE …

我有2个spark RDD,dataRDD和newPairDataRDD,它们用于spark SQL查询.当我的应用程序初始化时,dataRDD将被初始化.一个指定的hbase实体中的所有数据都将存储到dataRDD.

当客户端的SQL查询到来时,我的APP将获得所有新的更新并插入newPairDataRDD.dataRDD联合newPairDataRDD并在spark SQL上下文中注册为表.

我在dataRDD中找到了0条记录,在newPairDataRDD中找到了1条新的插入记录.工会需要4秒钟.那太慢了

我认为这是不合理的.谁知道怎么做得更快？谢谢简单的代码如下

    // Step1: load all data from hbase to dataRDD when initial, this only run once. 
    JavaPairRDD<String, Row>  dataRDD= getAllBaseDataToJavaRDD();
    dataRDD.cache();
    dataRDD.persist(StorageLevel.MEMORY_ONLY());
    logger.info(dataRDD.count());

    // Step2: when spark sql query coming, load latest updated and inserted data from db to newPairDataRDD

    JavaPairRDD<String, Row> newPairDataRDD = getUpdateOrInstertBaseDataToJavaRDD();
    // Step3: if count>0 do union and reduce

       if(newPairDataRDD.count() > 0) {

        JavaPairRDD<String, Row> unionedRDD =dataRDD.union(newPairDataRDD);

    // if data was updated in DB, need to delete the old version …