代码之家 › 专栏 › 技术社区 › Karthick

使用spark中的下推查询,如何在spark HBASE(BIGSQL as SQL engine)中获得并行性?

bigsql hbase parallel-processing dataframe apache-spark

Karthick · 技术社区 · 8 年前

在Spark中,数据库的SQL引擎对PushdownQuery进行处理,并根据处理结果构造数据帧。所以,spark查询这个查询的结果。

val pushdownQuery = """(SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah )"""

val dbDataFrame = spark.read.format("jdbc")
.option("url", url).option("dbtable", pushdownQuery).option("driver", driver)
.option("numPartitions", 4)
.option("partitionColumn", "COUNTRY_CODE")
.load()

我可以从另一个参考资料中看出( https://dzone.com/articles/how-apache-spark-makes-your-slow-mysql-queries-10x )在spark-mysql中,下推查询的并行性是通过基于参数numPartitions&partitionColumn触发多个查询来实现的。这与sqoop的分布方式非常相似。例如上面给定的参数示例numPartitions=4;partitionColumn=COUNTRY_CODE,在我们的表COUNTRY_CODE中,值范围为(000999)。

生成4个查询;向DB触发,并根据这些查询的结果构造dataframe(在本例中并行度为4)。

Q1 : SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah WHERE COUNTRY_CODE >= 000 AND COUNTRY_CODE <= 250
Q2 : SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah WHERE COUNTRY_CODE > 250 AND COUNTRY_CODE  <= 500
Q3 : SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah WHERE COUNTRY_CODE > 500 AND COUNTRY_CODE  <= 750
Q4 : SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah WHERE COUNTRY_CODE > 750 AND COUNTRY_CODE  <= 999

我现在的问题是,如何在spark(2.1版)+hbase(查询引擎-BIGSQL)中使用此方法实现并行性?现在还没有给我平行度。 spark hbase桥的驱动程序需要更新吗?或者spark需要这么做?或者是什么样的改变能帮助它做到这一点? 有些方向对我有帮助。谢谢您!

1 回复 | 直到 8 年前

Mansoor Baba Shaik 8 年前

为了获得最佳性能,我建议使用--num executors 4和--executor cores 1启动spark作业,因为jdbc连接是单线程的,每个查询一个内核上运行一个任务。通过进行此配置更改,当作业运行时,可以观察到并行运行的任务,这是每个执行器中正在使用的核心任务。

请改用以下函数:

val connectionProperties: Properties = new Properties
connectionProperties.put("user", "xxxx")
connectionProperties.put("password", "xxxx")
connectionProperties.put("fetchsize", "10000") //fetches 10000 records at once per task
connectionProperties.put("driver", "com.mysql.jdbc.Driver")
connectionProperties

val pushdownQuery = """(SELECT DISTINCT(FLIGHT_NUMBER) blah blah blah ) tbl_alias"""

val dbDataFrame = spark.read.jdbc(url, pushdownQuery, "COUNTRY_CODE", 0L, 4L, 4, connectionProperties)

参考 https://spark.apache.org/docs/2.3.0/api/scala/index.html#org.apache.spark.sql.DataFrameReader@jdbc(url:String,table:String,columnName:String,lowerBound:Long,upperBound:Long,numPartitions:Int,connectionProperties:java.util.Properties):org.apache.spark.sql.DataFrame