How to connect HBase and Spark using Python?

by

I found this comment by one of the makers of hbase-spark, which seems to suggest there is a way to use PySpark to query HBase using Spark SQL.

And indeed, the pattern described here can be applied to query HBase with Spark SQL using PySpark, as the following example shows:

from pyspark import SparkContext
from pyspark.sql import SQLContext

sc = SparkContext()
sqlc = SQLContext(sc)

data_source_format="org.apache.hadoop.hbase.spark"

df = sc.parallelize([('a', '1.0'), ('b', '2.0')]).toDF(schema=['col0', 'col1'])

# ''.join(string.split()) in order to write a multi-line JSON string here.
catalog = ''.join("""{
    "table":{"namespace":"default", "name":"testtable"},
    "rowkey":"key",
    "columns":{
        "col0":{"cf":"rowkey", "col":"key", "type":"string"},
        "col1":{"cf":"cf", "col":"col1", "type":"string"}
    }
}""".split())


# Writing
df.write\
.options(catalog=catalog)\  # alternatively: .option('catalog', catalog)
.format(data_source_format)\
.save()

# Reading
df = sqlc.read\
.options(catalog=catalog)\
.format(data_source_format)\
.load()

I’ve tried hbase-spark-1.2.0-cdh5.7.0.jar (as distributed by Cloudera) for this, but ran into trouble (org.apache.hadoop.hbase.spark.DefaultSource does not allow create table as select when writing, java.util.NoSuchElementException: None.get when reading). As it turns out, the present version of CDH does not include the changes to hbase-spark that allow Spark SQL-HBase integration.

What does work for me is the shc Spark package, found here. The only change I had to make to the above script is to change:

data_source_format="org.apache.spark.sql.execution.datasources.hbase"

Here’s how I submit the above script on my CDH cluster, following the example from the shc README:

spark-submit --packages com.hortonworks:shc:1.0.0-1.6-s_2.10 --repositories http://repo.hortonworks.com/content/groups/public/ --files /opt/cloudera/parcels/CDH/lib/hbase/conf/hbase-site.xml example.py

Most of the work on shc seems to already be merged into the hbase-spark module of HBase, for release in version 2.0. With that, Spark SQL querying of HBase is possible using the above-mentioned pattern (see: https://hbase.apache.org/book.html#_sparksql_dataframes for details). My example above shows what it looks like for PySpark users.

Finally, a caveat: my example data above has only strings. Python data conversion is not supported by shc, so I had problems with integers and floats not showing up in HBase or with weird values.

More Related Contents:

  1. Spark Equivalent of IF Then ELSE
  2. How to add a constant column in a Spark DataFrame?
  3. Spark Dataframe distinguish columns with duplicated name
  4. How do I add a new column to a Spark DataFrame (using PySpark)?
  5. How to perform union on two DataFrames with different amounts of columns in spark?
  6. Does spark predicate pushdown work with JDBC?
  7. How to change dataframe column names in pyspark?
  8. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  9. Explode in PySpark
  10. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  11. Passing a data frame column and external list to udf under withColumn
  12. Updating a dataframe column in spark
  13. Dividing complex rows of dataframe to simple rows in Pyspark
  14. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  15. Why is Apache-Spark – Python so slow locally as compared to pandas?
  16. Spark add new column to dataframe with value from previous row
  17. PySpark create new column with mapping from a dict
  18. Pyspark: Replacing value in a column by searching a dictionary
  19. Create Spark DataFrame. Can not infer schema for type
  20. Pivot String column on Pyspark Dataframe
  21. GroupBy column and filter rows with maximum value in Pyspark
  22. Using UDF ignores condition in when
  23. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  24. Concatenate two PySpark dataframes
  25. Filtering DataFrame using the length of a column
  26. Add column sum as new column in PySpark dataframe
  27. How to drop all columns with null values in a PySpark DataFrame?
  28. Pyspark changing type of column from date to string
  29. PySpark: compute row maximum of the subset of columns and add to an exisiting dataframe
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment