Export Data From Apache Phoenix
Once in a while, you come across technology from your past that triggers certain memories and, combined with an interesting enough challenge, has you commit to writing (seemingly) ancient code. For me, this was a customer's ask to help export data from an on-premises Apache HBase setup (using the Cloudera Distribution for Hadoop, in short CDH, aka CDP) with its SQL frontend based on Apache Phoenix. I have a history with HBase and had the proverbial famous last words in my head: How hard can that be?
Well, the tl;dr is, if you know what you are doing, it is just a few lines of code…, but if you do not know, it may take days to figure out. This post is to possibly save you some time and allow future me to relearn from past me. The code for this project lives in GitHub and contains a README.md file with all the technical information so that we can concentrate on the important takeaways here.
Requirements
Before we go any deeper, here are the requirements for this project:
- Data is stored in an older, on-premises version of Hadoop Distribution.
- Access to data is via Phoenix, which stores the SQL table in one or more columns (and column families) in HBase, using its own encoding formats.
- Bulk exporting from such a table should happen through a distributed MapReduce job at scale.
- The exported data is written to Parquet-formatted output files, which can then be loaded into Delta tables, for instance.
Why are we using MapReduce and not the JDBC driver available for Phoenix via a Spark job, for example? The reason is scalability, as reading terabytes of data via JDBC is cumbersome and would require careful query planning to achieve proper task distribution.
Setting up a test rig
Again, I have set up HBase locally and on clusters many times, even recently packaging this into a Docker container. The main reason is to control the environment (and, of course, protect the operating system from unnecessary pollution). However, since I also needed to run MapReduce jobs against HBase, I decided not to use the container setup in favor of setting up everything locally on my MacOS. Long story short, I spent hours fighting the weirdest issues caused by MacOS not being a derivative of Linux:
- While everything ran against
localhostas the hostname, I found in some arbitrary comment on Stack Overflow that for MacOS, you have to add the actual machine’s name to the/etc/hostsfile as an alternative tolocalhostin the line starting with127.0.0.1. Otherwise, the connections would fail with anUnresolvedAddressException. And even then, this would only work for direct connections, not from within the MapReduce job. - The resulting JAR contains the classes and
META-INFfiles needed for dependent libraries, including multiple LICENSE files. In MacOS, the file name casing causes the JAR to not work, as during the temporary staging of these files, it collides with other files of that name. You can get around this like somvn package && zip -d target/phoenix-exporter-mapreduce-job-1.0-SNAPSHOT-jar-with-dependencies.jar META-INF/LICENSE, but that is not pretty at best. - Getting the Java SDK and Python versions right is another challenge that may take too long. I had issues where the Python-based CLI scripts for Phoenix failed with both Python 2 and 3 installed.
The repo contains the full documentation on how to set up HBase, Phoenix, and Hadoop locally, purely based on tarballs matching the Cloudera CDH version the export job was meant to run in. After starting HBase like so,
$ bin/stop-hbase.sh && bin/start-hbase.shyou can access the Master WebUI by pointing a browser at http://localhost:16010, which should look like this:
Equally easy then is to open the Phoenix CLI tool with bin/sqlline.py (note that the linked README.md is showing the proper directories to choose):
Also, note that you can ignore the many WARNING level log statements for both HBase and Phoenix. These are notoriously noisy (although Zookeeper is the worst offender for this) and require a healthy amount of meh! attitude. The warnings often relate to the simple local setup where only some things are configured fully.
Finally, for MapReduce, follow the instructions in the repo running a test MapReduce job.
Setting up test tables
Now we need some test tables and data, which the repo also contains in the docs directory. Here is where I used the power of GenAI and instructed ChatGPT to generate a sample schema around a fictitious airline management system, along with 10 test records per table. The resulting SQL statements can be loaded into Phoenix using the CLI like so (assuming you are in the root directory called phoenix-exporter as explained in the README.md and the repo is cloned into mrjob inside of it):
$ bin/sqlline.py localhost ../mrjob/docs/schemas.sql
$ bin/sqlline.py localhost ../mrjob/docs/data.sql The scripts will finish quickly, and you can confirm both sides, that is Phoenix and HBase, using their CLI tools:
Tip: Use !outputformat vertical at the command prompt when running sqlline.py to have it emit the above vertically, as often the screen is too small to scroll horizontally.
Obviously, on the HBase side, the rows are serialized into byte arrays since HBase is a type-less system, and everything must be stored as bytes. The row key is a composite of three Phoenix columns, and, in addition, the column names present in the schemas are converted to shorter, space-saving numeric IDs (although enabling compression, which I omitted for the sake of simplicity, would mostly take care of repeated column names).
Get Lars George’s stories in your inbox
Join Medium for free to get updates from this writer.
How you serialize typed data for HBase is flexible, and ideally, you should use the supplied helper classes to access the data and let the necessary conversion alone. On top of that, some tables in the example are deliberately salted to achieve a better physical spread of consecutive row keys across the HBase region servers. All of this leads to the realization that reading the data directly from HBase is not viable (although possible).
Exporting data
Instead, it would be best to look for what the library that talks to HBase has to offer, and lo and behold, the Phoenix project supplies a class that can read its tables and provide a deserialized record for every row. Writing out the data in text format (say, JSON or CSV) is just too error-prone, so here we pick Apache Parquet, the prominent file format in big data and the basis for abstraction layers, such as Delta Lake. Again, we look toward the project and find that it provides the classes to write the table rows into Parquet files.
The only choice is how to provide the required output schema for Parquet to write the data properly. For that, Parquet teams up with Apache Avro, borrowing its schema definition-related classes and wrapping the Parquet output format into one that expects the schema to be supplied in Avro style.
I tried that approach first but eventually gave up as I encountered too many problems:
- Avro uses another schema definition language, so you must translate from Phoenix to Avro to Parquet, leaving many questions open.
- Since I wanted the code to infer the Parquet schema from the typed Phoenix one, using JSON as the definition language in Avro was just not right, and using the schema builder classes supplied by Avro (as they are pretty much undocumented) caused hours of wasted, unsuccessful trial and error attempts.
Ultimately, I removed the Avro dependency from the Maven manifest and tried it with plain Parquet instead. This required implementing the custom classes that convert the schema and data from Phoenix to Parquet.
In short, the custom WriteSupport class takes a deserialized Phoenix row in the form of a DBWritable subclass. This happens behind the scenes as part of the output format using the custom write support, which means int the Mapper implementation, we can simply emit the Phoenix-provided record instance.
There is also a helper class to retrieve the Phoenix schema of the exported table via an RPC call from the HBase coprocessors. The above write support implementation triggers the conversion of the list of PColumn instances to Parquet Type subclasses.
Results
What remains is running the actual job! Compiling and running the code is explained in the repo, letting us focus on the outcome. We start the job and ask it to export the above BOOKINGS table from Phoenix:
$ rm -fR /tmp/bookings && \
bin/hadoop jar \
../mrjob/target/phoenix-exporter-mapreduce-job-1.0-SNAPSHOT-jar-with-dependencies.jar \
BOOKINGS /tmp/bookings
...There is a lot emitted on the console running the job, including how the code is converting the table schema to the matching Parquet one:
...
2024-08-21 11:04:13,465 INFO hbase.PhoenixToParquetHelper: Skipping column: _SALT
2024-08-21 11:04:13,470 INFO hbase.PhoenixToParquetHelper: Data type: UNSIGNED_INT -> Converted type: required int32 BOOKING_ID - Nullable: false
2024-08-21 11:04:13,470 INFO hbase.PhoenixToParquetHelper: Data type: UNSIGNED_INT -> Converted type: required int32 FLIGHT_ID - Nullable: false
2024-08-21 11:04:13,470 INFO hbase.PhoenixToParquetHelper: Data type: UNSIGNED_INT -> Converted type: required int32 PASSENGER_ID - Nullable: false
2024-08-21 11:04:13,474 INFO hbase.PhoenixToParquetHelper: Data type: TIMESTAMP -> Converted type: optional int64 BOOKING_DATE (TIMESTAMP(MILLIS,true)) - Nullable: true
2024-08-21 11:04:13,474 INFO hbase.PhoenixToParquetHelper: Data type: VARCHAR -> Converted type: optional binary SEAT_NUMBER (STRING) - Nullable: true
2024-08-21 11:04:13,474 INFO hbase.PhoenixToParquetHelper: Data type: VARCHAR -> Converted type: optional binary STATUS (STRING) - Nullable: true
2024-08-21 11:04:13,483 INFO hbase.PhoenixToParquetHelper: Data type: DECIMAL -> Converted type: optional binary PRICE_PAID (DECIMAL(10,2)) - Nullable: true
2024-08-21 11:04:13,483 INFO hbase.PhoenixToParquetHelper: Data type: UNSIGNED_LONG -> Converted type: optional int64 POINTS_EARNED - Nullable: true
2024-08-21 11:04:13,483 INFO hbase.PhoenixToParquetHelper: Number of converted columns: 8
2024-08-21 11:04:13,484 INFO hbase.PhoenixToParquetWriteSupport: Schema: message row {
required int32 BOOKING_ID;
required int32 FLIGHT_ID;
required int32 PASSENGER_ID;
optional int64 BOOKING_DATE (TIMESTAMP(MILLIS,true));
optional binary SEAT_NUMBER (STRING);
optional binary STATUS (STRING);
optional binary PRICE_PAID (DECIMAL(10,2));
optional int64 POINTS_EARNED;
}
...At the end of the run, you should see it stating how many records it read and wrote, here 10 for each, as that is what we had in our test dataset:
...
2024-08-21 11:04:14,310 INFO mapred.LocalJobRunner: Finishing task: attempt_local676674579_0001_m_000009_0
2024-08-21 11:04:14,310 INFO mapred.LocalJobRunner: map task executor complete.
2024-08-21 11:04:14,318 INFO hadoop.ParquetFileReader: Initiating action with parallelism: 5
2024-08-21 11:04:15,197 INFO mapreduce.Job: Job job_local676674579_0001 completed successfully
2024-08-21 11:04:15,230 INFO mapreduce.Job: Counters: 15
File System Counters
FILE: Number of bytes read=2659854134
FILE: Number of bytes written=2686382072
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
Map-Reduce Framework
Map input records=10
Map output records=10
Input split bytes=5587
Spilled Records=0
Failed Shuffles=0
Merged Map outputs=0
GC time elapsed (ms)=5
Total committed heap usage (bytes)=2600468480
File Input Format Counters
Bytes Read=0
File Output Format Counters
Bytes Written=14906Since this table is salted, you also should note that we had a skipped column above, called _SALT, that tracks the prefix to spread the rows across the ten regions. We also had ten mappers, one for each region, emitting data into a separate Parquet file, which is what we need to parallelize the work:
$ ls /tmp/bookings/
_SUCCESS part-m-00000.snappy.parquet part-m-00003.snappy.parquet part-m-00006.snappy.parquet part-m-00009.snappy.parquet
_common_metadata part-m-00001.snappy.parquet part-m-00004.snappy.parquet part-m-00007.snappy.parquet
_metadata part-m-00002.snappy.parquet part-m-00005.snappy.parquet part-m-00008.snappy.parquet Finally, we can use the Parquet-supplied helper tool to dump the content of one of the Parquet files. If you look for the value: rows, you can see our test data made it into Parquet.
$ bin/hadoop jar ../parquet-tools-1.11.2.jar dump /tmp/bookings/part-m-00000.snappy.parquet
...
INT32 BOOKING_ID
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:0 V:5
INT32 FLIGHT_ID
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:0 V:5
INT32 PASSENGER_ID
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:0 V:5
INT64 BOOKING_DATE
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:1 V:2024-08-20T14:28:43.133+0000
BINARY SEAT_NUMBER
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:1 V:1A
BINARY STATUS
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:1 V:Canceled
BINARY PRICE_PAID
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:1 V:559.99
INT64 POINTS_EARNED
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 1 ***
value 1: R:0 D:1 V:0From here, converting the files into a Delta table or importing the data using a data pipeline is easy.
The above works at scale, handles Phoenix tables as expected, and can also be enhanced to hand in a start_time, which would be passed on to the internal scan operation only to read new rows. Happy exporting everyone!
