Overview

OLTP and operational analytics for Apache Hadoop

Apache Phoenix enables OLTP and operational analytics in Hadoop for low-latency applications by combining:

• the power of standard SQL and JDBC APIs, with full ACID transaction capabilities
• the flexibility of late-bound, schema-on-read capabilities from the NoSQL world by leveraging HBase as its backing store

Apache Phoenix is fully integrated with other Hadoop products such as Spark, Hive, Pig, Flume, and MapReduce.

Who is using Apache Phoenix? Read more here.

Mission

Become the trusted data platform for OLTP and operational analytics for Hadoop through well-defined, industry-standard APIs.

Quick Start

Tired of reading and just want to get started? Take a look at our FAQs, listen to the Apache Phoenix talk from Hadoop Summit 2015, review the overview presentation, and jump to our quick start guide here.

SQL Support

Apache Phoenix takes your SQL query, compiles it into a series of HBase scans, and orchestrates those scans to produce regular JDBC result sets. Direct use of the HBase API, along with coprocessors and custom filters, results in performance on the order of milliseconds for small queries, or seconds for tens of millions of rows.

To see a complete list of what is supported, go to our language reference. All standard SQL query constructs are supported, including SELECT, FROM, WHERE, GROUP BY, HAVING, ORDER BY, etc. It also supports a full set of DML commands, as well as table creation and versioned incremental alterations through DDL commands.

Here is a list of what is currently not supported:

• Relational operators: INTERSECT, MINUS
• Miscellaneous built-in functions: these are easy to add — read this blog for step-by-step instructions.

Connection

Use JDBC to get a connection to an HBase cluster:

Connection conn = DriverManager.getConnection("jdbc:phoenix:server1,server2:3333", props);

Where props are optional properties that may include Phoenix and HBase configuration values.

The JDBC connection string is composed as:

jdbc:phoenix[:<zookeeper quorum>[:<port number>[:<root node>[:<principal>[:<keytab file>]]]]]

For omitted parts, values are taken from hbase-site.xml (hbase.zookeeper.quorum, hbase.zookeeper.property.clientPort, and zookeeper.znode.parent).

The optional principal and keytab file may be used to connect to a Kerberos-secured cluster. If only principal is specified, each distinct user gets a dedicated HBase connection (HConnection), allowing multiple different connections with different configuration properties on the same JVM.

For example, for longer-running queries:

Connection conn = DriverManager.getConnection("jdbc:phoenix:my_server:longRunning", longRunningProps);

And for shorter-running queries:

Connection conn = DriverManager.getConnection("jdbc:phoenix:my_server:shortRunning", shortRunningProps);

See the relevant FAQ entry for example URLs.

Phoenix also supports connecting to HBase without ZooKeeper.

Transactions

To enable full ACID transactions (beta in 4.7.0), set phoenix.transactions.enabled=true. In this case, you also need to run the transaction manager included in the distribution. Once enabled, a table may optionally be declared as transactional (see transactions).

Commits over transactional tables are all-or-none: either all data is committed (including secondary index updates) or none is committed (and an exception is thrown). Both cross-table and cross-row transactions are supported. Transactional tables also see their own uncommitted data when querying. An optimistic concurrency model is used to detect row-level conflicts with first-commit-wins semantics.

Non-transactional tables have no guarantees beyond HBase row-level atomicity (see HBase ACID semantics). Also, non-transactional tables do not see updates until commit occurs.

Phoenix DML commands (UPSERT VALUES, UPSERT SELECT, DELETE) batch pending changes on the client side. Changes are sent to server on commit and discarded on rollback. If auto-commit is enabled, Phoenix will execute the entire DML command server-side via coprocessors whenever possible for better performance.

Timestamps

Most applications let HBase manage timestamps. In cases where timestamps must be controlled, the CurrentSCN property can be set at connection time to control timestamps for DDL, DML, and queries. This also enables snapshot queries against prior row values because Phoenix uses this property as scan max timestamp.

Timestamps cannot be controlled for transactional tables. Instead, the transaction manager assigns timestamps that become HBase cell timestamps on commit. They still correspond to wall-clock time, but are multiplied by 1,000,000 to ensure enough granularity for uniqueness across the cluster.

Schema

Apache Phoenix supports table creation and versioned incremental alterations through DDL commands. Table metadata is stored in an HBase table and versioned, so snapshot queries over prior versions automatically use the correct schema.

A Phoenix table can be created through CREATE TABLE and can either be:

1. Built from scratch: HBase table and column families are created automatically.
2. Mapped to an existing HBase table: either as a read-write TABLE or a read-only VIEW, with the caveat that row key/key-value binary representation must match Phoenix data types (see Data Types).
   • For a read-write TABLE, column families are created automatically if absent. An empty key value is added to the first column family of each existing row to minimize projection size for queries.
   • For a read-only VIEW, all column families must already exist. The only change is adding Phoenix coprocessors for query processing. The primary use case is transferring existing data into Phoenix; DML is not allowed on a VIEW, and query performance may be lower than with a TABLE.

All schema is versioned (up to 1000 versions kept). Snapshot queries over older data pick up the correct schema based on connection time via CurrentSCN.

Altering

A Phoenix table may be altered via ALTER TABLE. When a SQL statement references a table, Phoenix checks with the server by default to ensure metadata and statistics are up to date.

If table structure is known to be stable, this RPC may be unnecessary. UPDATE_CACHE_FREQUENCY (added in 4.7) lets users define how often the server is checked for metadata/statistics updates. Possible values: ALWAYS (default), NEVER, or a millisecond value.

For example, this DDL creates table FOO and tells clients to check for updates every 15 minutes:

CREATE TABLE FOO (k BIGINT PRIMARY KEY, v VARCHAR) UPDATE_CACHE_FREQUENCY=900000;

Views

Phoenix supports updatable views on top of tables, with the unique capability of adding columns by leveraging HBase schemaless behavior. All views share the same underlying HBase table and may be indexed independently. Read more here.

Multi-tenancy

Built on top of view support, Phoenix also supports multi-tenancy. As with views, a multi-tenant view can add columns defined solely for that user.

Schema at read time

Another schema-related feature allows columns to be defined dynamically at query time. This is useful when not all columns are known at create time. More details here.

Mapping to an Existing HBase Table

Phoenix supports mapping to an existing HBase table through CREATE TABLE and CREATE VIEW. In both cases, HBase metadata remains as-is, except that with CREATE TABLE, KEEP_DELETED_CELLS is enabled so flashback queries work correctly.

For CREATE TABLE, missing HBase metadata (table/column families) is created if needed. Table and column family names are case-sensitive at HBase level; Phoenix uppercases names by default. To preserve case sensitivity, wrap names in double quotes:

CREATE VIEW "MyTable" ("a".ID VARCHAR PRIMARY KEY);

For CREATE TABLE, an empty key value is added per row so queries behave as expected without projecting all columns during scans. For CREATE VIEW, this is not done and no HBase metadata is created. Existing HBase metadata must match DDL metadata, or ERROR 505 (42000): Table is read only is thrown.

Another caveat: bytes serialized in HBase must match Phoenix serialization expectations.

• For VARCHAR, CHAR, and UNSIGNED_* types, Phoenix uses HBase Bytes utility methods.
• CHAR expects only single-byte characters.
• UNSIGNED_* expects non-negative values.

Composite row keys are formed by concatenating values, with a zero byte separator after variable-length types. For more on type system details, see Data Types.

Salting

Tables can be declared salted to avoid HBase region hotspotting. Declare a salt bucket count and Phoenix manages salting transparently. See details here, and write-throughput comparison here.

APIs

Catalog metadata (tables, columns, primary keys, and types) can be retrieved through Java SQL metadata interfaces: DatabaseMetaData, ParameterMetaData, and ResultSetMetaData.

For schema/table/column retrieval via DatabaseMetaData, schema pattern, table pattern, and column pattern are LIKE-style expressions (% and _, escaped by \).

In metadata APIs, table catalog argument is used to filter by tenant ID for multi-tenant tables.

Quick Start

What is this new Phoenix thing I've been hearing about?

Phoenix is an open source SQL skin for HBase. You use the standard JDBC APIs instead of the regular HBase client APIs to create tables, insert data, and query your HBase data.

Doesn't putting an extra layer between my application and HBase just slow things down?

Actually, no. Phoenix achieves as good or likely better performance than if you hand-coded it yourself (not to mention with a heck of a lot less code) by:

• compiling your SQL queries to native HBase scans
• determining the optimal start and stop for your scan key
• orchestrating the parallel execution of your scans
• bringing the computation to the data by
  • pushing the predicates in your where clause to a server-side filter
  • executing aggregate queries through server-side hooks (called co-processors)
• secondary indexes to improve performance for queries on non row key columns
• stats gathering to improve parallelization and guide choices between optimizations
• skip scan filter to optimize IN, LIKE, and OR queries
• optional salting of row keys to evenly distribute write load

Ok, so it's fast. But why SQL? It's so 1970s

Well, that's kind of the point: give folks something with which they're already familiar. What better way to spur the adoption of HBase? On top of that, using JDBC and SQL:

• Reduces the amount of code users need to write
• Allows for performance optimizations transparent to the user
• Opens the door for leveraging and integrating lots of existing tooling

But how can SQL support my favorite HBase technique of x,y,z

Didn't make it to the last HBase Meetup did you? SQL is just a way of expressing what you want to get not how you want to get it. Check out my presentation for various existing and to-be-done Phoenix features to support your favorite HBase trick. Have ideas of your own? We'd love to hear about them: file an issue for us and/or join our mailing list.

Blah, blah, blah - I just want to get started!

Ok, great! Just follow our install instructions:

• download and expand our installation binary tar corresponding to your HBase version
• copy the phoenix server jar into the lib directory of every region server and master
• restart HBase
• add the phoenix client jar to the classpath of your JDBC client or application
• We have detailed instructions for setting up SQuirreL SQL as your SQL client

I don't want to download and setup anything else!

Ok, fair enough - you can create your own SQL scripts and execute them using our command line tools instead. Let's walk through an example now. Begin by navigating to the bin/ directory of your Phoenix install location.

CREATE TABLE IF NOT EXISTS us_population (
  state CHAR(2) NOT NULL,
  city VARCHAR NOT NULL,
  population BIGINT
  CONSTRAINT my_pk PRIMARY KEY (state, city)
);

NY,New York,8143197
CA,Los Angeles,3844829
IL,Chicago,2842518
TX,Houston,2016582
PA,Philadelphia,1463281
AZ,Phoenix,1461575
TX,San Antonio,1256509
CA,San Diego,1255540
TX,Dallas,1213825
CA,San Jose,912332

./psql.py <your_zookeeper_quorum> us_population.sql us_population.csv

./sqlline.py <your_zookeeper_quorum>

SELECT state as "State",count(city) as "City Count",sum(population) as "Population Sum"
FROM us_population
GROUP BY state
ORDER BY sum(population) DESC;

Congratulations! You've just created your first Phoenix table, inserted data into it, and executed an aggregate query with just a few lines of code in 15 minutes or less!

Big deal - 10 rows! What else you got?

Ok, ok - tough crowd. Check out our bin/performance.py script to create as many rows as you want, for any schema you come up with, and run timed queries against it.

Why is it called Phoenix anyway? Did some other project crash and burn and this is the next generation?

I'm sorry, but we're out of time and space, so we'll have to answer that next time!

FAQ

Questions answered in this page:

• I want to get started. Is there a Phoenix Hello World?
• What is the Phoenix JDBC URL syntax?
• Is there a way to bulk load in Phoenix?
• How I map Phoenix table to an existing HBase table?
• Are there any tips for optimizing Phoenix?
• How do I create Secondary Index on a table?
• Why isn't my secondary index being used?
• How fast is Phoenix? Why is it so fast?
• How do I connect to secure HBase cluster?
• What HBase and Hadoop versions are supported?
• Can phoenix work on tables with arbitrary timestamp as flexible as HBase API?
• Why isn't my query doing a RANGE SCAN?
• Should I pool Phoenix JDBC Connections?
• Why does Phoenix add an empty or dummy KeyValue when doing an upsert?

I want to get started. Is there a Phoenix Hello World?

Pre-requisite: Download and install the latest Phoenix.

Using console

$ sqlline.py [zookeeper quorum hosts]

create table test (mykey integer not null primary key, mycolumn varchar);
upsert into test values (1,'Hello');
upsert into test values (2,'World!');
select * from test;

+-------+------------+
| MYKEY |  MYCOLUMN  |
+-------+------------+
| 1     | Hello      |
| 2     | World!     |
+-------+------------+

Using Java

Create test.java file with the following content:

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.PreparedStatement;
import java.sql.Statement;

public class test {

	public static void main(String[] args) throws SQLException {
		Statement stmt = null;
		ResultSet rset = null;

		Connection con = DriverManager.getConnection("jdbc:phoenix:[zookeeper quorum hosts]");
		stmt = con.createStatement();

		stmt.executeUpdate("create table test (mykey integer not null primary key, mycolumn varchar)");
		stmt.executeUpdate("upsert into test values (1,'Hello')");
		stmt.executeUpdate("upsert into test values (2,'World!')");
		con.commit();

		PreparedStatement statement = con.prepareStatement("select * from test");
		rset = statement.executeQuery();
		while (rset.next()) {
			System.out.println(rset.getString("mycolumn"));
		}
		statement.close();
		con.close();
	}
}

Compile and execute on command line

$ javac test.java
$ java -cp "../phoenix-[version]-client.jar:." test

You should get the following output

Hello
World!

What is the Phoenix JDBC URL syntax?

Thick Driver

See Using the Phoenix JDBC Driver for a more up-to-date description

The Phoenix (Thick) Driver JDBC URL syntax is as follows (where elements in square brackets are optional):

jdbc:phoenix:[comma-separated ZooKeeper Quorum Hosts [: ZK port [:hbase root znode [:kerberos_principal [:path to kerberos keytab] ] ] ]

The simplest URL is:

jdbc:phoenix

Whereas the most complicated URL is:

jdbc:phoenix:zookeeper1.domain,zookeeper2.domain,zookeeper3.domain:2181:/hbase-1:phoenix@EXAMPLE.COM:/etc/security/keytabs/phoenix.keytab

Please note that each optional element in the URL requires all previous optional elements. For example, to specify the HBase root ZNode, the ZooKeeper port must also be specified.

See also Connection String.

Thin Driver

The Phoenix Thin Driver (used with the Phoenix Query Server) JDBC URL syntax is as follows:

jdbc:phoenix:thin:[key=value[;key=value...]]

There are a number of keys exposed for client-use. The most commonly-used keys are: url and serialization. The url key is required to interact with the Phoenix Query Server.

The simplest URL is:

jdbc:phoenix:thin:url=http://localhost:8765

Where as very complicated URL is:

jdbc:phoenix:thin:url=http://queryserver.domain:8765;serialization=PROTOBUF;authentication=SPENGO;principal=phoenix@EXAMPLE.COM;keytab=/etc/security/keytabs/phoenix.keytab

Please refer to the Apache Avatica documentation for a full list of supported options in the Thin client JDBC URL, or see the Query Server documentation

Is there a way to bulk load in Phoenix?

Map Reduce

See the example here

CSV

CSV data can be bulk loaded with built in utility named psql. Typical upsert rates are 20K - 50K rows per second (depends on how wide are the rows).

Usage example:

• Create table using psql:

  $ psql.py [zookeeper] ../examples/web_stat.sql

• Upsert CSV bulk data:

  $ psql.py [zookeeper] ../examples/web_stat.csv

How I map Phoenix table to an existing HBase table?

You can create both a Phoenix table or view through the CREATE TABLE/CREATE VIEW DDL statement on a pre-existing HBase table. In both cases, we'll leave the HBase metadata as-is. For CREATE TABLE, we'll create any metadata (table, column families) that doesn't already exist. We'll also add an empty key value for each row so that queries behave as expected (without requiring all columns to be projected during scans).

The other caveat is that the way the bytes were serialized must match the way the bytes are serialized by Phoenix. For VARCHAR, CHAR, and UNSIGNED_* types, we use the HBase Bytes methods. The CHAR type expects only single-byte characters and the UNSIGNED types expect values greater than or equal to zero. For signed types (TINYINT, SMALLINT, INTEGER and BIGINT), Phoenix will flip the first bit so that negative values will sort before positive values. Because HBase sorts row keys in lexicographical order and negative value's first bit is 1 while positive 0 so that negative value is 'greater than' positive value if we don't flip the first bit. So if you stored integers by HBase native API and want to access them by Phoenix, make sure that all your data types are UNSIGNED types.

Our composite row keys are formed by simply concatenating the values together, with a zero byte character used as a separator after a variable length type.

If you create an HBase table like this:

create 't1', {NAME => 'f1', VERSIONS => 5}

then you have an HBase table with a name of t1 and a column family with a name of f1. Remember, in HBase, you don't model the possible KeyValues or the structure of the row key. This is the information you specify in Phoenix above and beyond the table and column family.

So in Phoenix, you'd create a view like this:

CREATE VIEW "t1" ( pk VARCHAR PRIMARY KEY, "f1".val VARCHAR )

The pk column declares that your row key is a VARCHAR (i.e. a string) while the "f1".val column declares that your HBase table will contain KeyValues with a column family and column qualifier of "f1":VAL and that their value will be a VARCHAR.

Note that you don't need the double quotes if you create your HBase table with all caps names (since this is how Phoenix normalizes strings, by upper casing them). For example, with:

create 'T1', {NAME => 'F1', VERSIONS => 5}

you could create this Phoenix view:

CREATE VIEW t1 ( pk VARCHAR PRIMARY KEY, f1.val VARCHAR )

Or if you're creating new HBase tables, just let Phoenix do everything for you like this (No need to use the HBase shell at all.):

CREATE TABLE t1 ( pk VARCHAR PRIMARY KEY, val VARCHAR )

Are there any tips for optimizing Phoenix?

• Use Salting to increase read/write performance
  Salting can significantly increase read/write performance by pre-splitting the data into multiple regions. Although Salting will yield better performance in most scenarios.
  Example:

  CREATE TABLE TEST (HOST VARCHAR NOT NULL PRIMARY KEY, DESCRIPTION VARCHAR) SALT_BUCKETS=16

  Note: Ideally for a 16 region server cluster with quad-core CPUs, choose salt buckets between 32-64 for optimal performance.

• Pre-split table
  Salting does automatic table splitting but in case you want to exactly control where table split occurs with out adding extra byte or change row key order then you can pre-split a table.
  Example:

  CREATE TABLE TEST (HOST VARCHAR NOT NULL PRIMARY KEY, DESCRIPTION VARCHAR) SPLIT ON ('CS','EU','NA')

• Use multiple column families
  Column family contains related data in separate files. If you query use selected columns then it make sense to group those columns together in a column family to improve read performance.
  Example:
  Following create table DDL will create two column faimiles A and B.

  CREATE TABLE TEST (MYKEY VARCHAR NOT NULL PRIMARY KEY, A.COL1 VARCHAR, A.COL2 VARCHAR, B.COL3 VARCHAR)

• Use compression
  On disk compression improves performance on large tables
  Example:

  CREATE TABLE TEST (HOST VARCHAR NOT NULL PRIMARY KEY, DESCRIPTION VARCHAR) COMPRESSION='GZ'

• Create indexes See How do I connect to secure HBase cluster?

• Optimize cluster parameters See https://hbase.apache.org/docs/performance

• Optimize Phoenix parameters See Configuration

How do I create Secondary Index on a table?

Starting with Phoenix version 2.1, Phoenix supports index over mutable and immutable data. Note that Phoenix 2.0.x only supports Index over immutable data. Index write performance index with immutable table is slightly faster than mutable table however data in immutable table cannot be updated.

Example:

• Create table
  Immutable table: create table test (mykey varchar primary key, col1 varchar, col2 varchar) IMMUTABLE_ROWS=true;
  Mutable table: create table test (mykey varchar primary key, col1 varchar, col2 varchar);
• Creating index on col2
  create index idx on test (col2)
• Creating index on col1 and a covered index on col2
  create index idx on test (col1) include (col2)
  Upsert rows in this test table and Phoenix query optimizer will choose correct index to use. You can see in explain plan if Phoenix is using the index table. You can also give a hint in Phoenix query to use a specific index.

See Secondary Indexing for further information

Why isn't my secondary index being used?

The secondary index won't be used unless all columns used in the query are in it ( as indexed or covered columns). All columns making up the primary key of the data table will automatically be included in the index.

Example: DDL create table usertable (id varchar primary key, firstname varchar, lastname varchar); create index idx_name on usertable (firstname);

Query: DDL select id, firstname, lastname from usertable where firstname = 'foo';

Index would not be used in this case as lastname is not part of indexed or covered column. This can be verified by looking at the explain plan. To fix this create index that has either lastname part of index or covered column. Example: create idx_name on usertable (firstname) include (lastname);

You can force Phoenix to use secondary for uncovered columns by specifying an index hint

How fast is Phoenix? Why is it so fast?

Phoenix is fast. Full table scan of 100M rows usually completes in 20 seconds (narrow table on a medium sized cluster). This time come down to few milliseconds if query contains filter on key columns. For filters on non-key columns or non-leading key columns, you can add index on these columns which leads to performance equivalent to filtering on key column by making copy of table with indexed column(s) part of key.

Why is Phoenix fast even when doing full scan:

1. Phoenix chunks up your query using the region boundaries and runs them in parallel on the client using a configurable number of threads.
2. The aggregation will be done in a coprocessor on the server-side, collapsing the amount of data that gets returned back to the client rather than returning it all.

How do I connect to secure HBase cluster?

Specify the principal and corresponding keytab in the JDBC URL as show above. For ancient Phoenix versions heck out the excellent post by Anil Gupta

What HBase and Hadoop versions are supported?

Phoenix 4.x supports HBase 1.x running on Hadoop 2

Phoenix 5.x supports HBase 2.x running on Hadoop 3

See the release notes and BUILDING in recent releases for the exact versions supported, and on how to build Phoenix for specific HBase and Hadoop versions

Can phoenix work on tables with arbitrary timestamp as flexible as HBase API?

By default, Phoenix lets HBase manage the timestamps and just shows you the latest values for everything. However, Phoenix also allows arbitrary timestamps to be supplied by the user. To do that you'd specify a CurrentSCN at connection time, like this:

Properties props = new Properties();
props.setProperty("CurrentSCN", Long.toString(ts));
Connection conn = DriverManager.connect(myUrl, props);

conn.createStatement().execute("UPSERT INTO myTable VALUES ('a')");
conn.commit();

The above is equivalent to doing this with the HBase API:

myTable.put(Bytes.toBytes('a'), ts);

By specifying a CurrentSCN, you're telling Phoenix that you want everything for that connection to be done at that timestamp. Note that this applies to queries done on the connection as well - for example, a query over myTable above would not see the data it just upserted, since it only sees data that was created before its CurrentSCN property. This provides a way of doing snapshot, flashback, or point-in-time queries.

Keep in mind that creating a new connection is not an expensive operation. The same underlying HConnection is used for all connections to the same cluster, so it's more or less like instantiating a few objects.

Why isn't my query doing a RANGE SCAN?

CREATE TABLE TEST (
  pk1 char(1) not null,
  pk2 char(1) not null,
  pk3 char(1) not null,
  non-pk varchar,
  CONSTRAINT PK PRIMARY KEY(pk1, pk2, pk3)
);

RANGE SCAN means that only a subset of the rows in your table will be scanned over. This occurs if you use one or more leading columns from your primary key constraint. Query that is not filtering on leading PK columns ex. select * from test where pk2='x' and pk3='y'; will result in full scan whereas the following query will result in range scan select * from test where pk1='x' and pk2='y';. Note that you can add a secondary index on your pk2 and pk3 columns and that would cause a range scan to be done for the first query (over the index table).

DEGENERATE SCAN means that a query can't possibly return any rows. If we can determine that at compile time, then we don't bother to even run the scan.

FULL SCAN means that all rows of the table will be scanned over (potentially with a filter applied if you have a WHERE clause)

SKIP SCAN means that either a subset or all rows in your table will be scanned over, however it will skip large groups of rows depending on the conditions in your filter. See this blog for more detail. We don't do a SKIP SCAN if you have no filter on the leading primary key columns, but you can force a SKIP SCAN by using the /*+ SKIP_SCAN */ hint. Under some conditions, namely when the cardinality of your leading primary key columns is low, it will be more efficient than a FULL SCAN.

Should I pool Phoenix JDBC Connections?

No, it is not necessary to pool Phoenix JDBC Connections.

Phoenix's Connection objects are different from most other JDBC Connections due to the underlying HBase connection. The Phoenix Connection object is designed to be a thin object that is inexpensive to create. If Phoenix Connections are reused, it is possible that the underlying HBase connection is not always left in a healthy state by the previous user. It is better to create new Phoenix Connections to ensure that you avoid any potential issues.

Implementing pooling for Phoenix could be done simply by creating a delegate Connection that instantiates a new Phoenix connection when retrieved from the pool and then closes the connection when returning it to the pool (see PHOENIX-2388).

Why does Phoenix add an empty/dummy KeyValue when doing an upsert?

The empty or dummy KeyValue (with a column qualifier of _0) is needed to ensure that a given column is available for all rows.

As you may know, data is stored in HBase as KeyValues, meaning that the full row key is stored for each column value. This also implies that the row key is not stored at all unless there is at least one column stored.

Now consider JDBC row which has an integer primary key, and several columns which are all null. In order to be able to store the primary key, a KeyValue needs to be stored to show that the row is present at all. This column is represented by the empty column that you've noticed. This allows doing a SELECT * FROM TABLE and receiving records for all rows, even those whose non-pk columns are null.

The same issue comes up even if only one column is null for some (or all) records. A scan over Phoenix will include the empty column to ensure that rows that only consist of the primary key (and have null for all non-key columns) will be included in a scan result.

Building

Building the Main Phoenix Project

Phoenix consists of several subprojects.

The core project is phoenix, which depends on phoenix-thirdparty, phoenix-omid, and phoenix-tephra.

phoenix-queryserver and phoenix-connectors are optional packages that also depend on phoenix.

Check out the source and follow the build instructions in BUILDING.md (or README.md) in the repository root.

Using Phoenix in a Maven Project

Phoenix artifacts are published to Apache and Maven Central repositories. Add the dependency below to your pom.xml:

<dependencies>
  <dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-client-hbase-[hbase.profile]</artifactId>
    <version>[phoenix.version]</version>
  </dependency>
</dependencies>

Where:

• [phoenix.version] is the Phoenix release version (for example, 5.1.2 or 4.16.1)
• [hbase.profile] is the compatible HBase profile

See Downloads for supported release/profile combinations.

Branches

The main Phoenix project currently has two active branches.

• 4.x works with HBase 1 and Hadoop 2
• 5.x works with HBase 2 and Hadoop 3

See Downloads and BUILDING.md for exact version compatibility by release.

See also:

• Building Project Website
• How to Release

Client Classpath and JDBC URL

Using the Phoenix JDBC Driver

This page is about using the Phoenix thick client.

The thin client for Phoenix Query Server is described on its own page.

The Phoenix classpath

To use Phoenix, both the JDBC driver JAR and hbase-site.xml must be added to the application classpath.

Phoenix driver JAR

The Phoenix JDBC client is built on top of the HBase client, and has an unusually high number of dependencies. To make this manageable, Phoenix provides a single shaded uberjar that can be added to the classpath.

Phoenix uses some private and semi-public HBase APIs, which may change between HBase versions, and provides separate binary distributions for different HBase versions.

Choose the binary distribution or Maven artifact corresponding to the HBase version on your cluster.

<dependencies>
  <dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-client-embedded-hbase-[hbase.profile]</artifactId>
    <version>[phoenix.version]</version>
  </dependency>
</dependencies>

HBase / Hadoop configuration files

As Phoenix is built on top of the HBase client, it needs the HBase configuration files for correct operation. For some configurations, it may also need other Hadoop / HDFS config files like core-site.xml.

Download the correct hbase-site.xml (the client one, usually in /etc/hbase/conf) from the cluster, and copy it to a directory on the classpath. It is important to add the directory containing hbase-site.xml, and not the full file path, to the classpath.

Alternatively, package hbase-site.xml into the root directory of a JAR file and add that JAR to the classpath.

If hbase-site.xml changes on the cluster, make sure to copy the updated file to your application classpath.

For some development clusters that use default configuration Phoenix may work without this, but not having the correct hbase-site.xml on the classpath is almost guaranteed to cause problems.

The Phoenix JDBC URL

The Phoenix URL contains two main parts. The first describes the connection to HBase; the second specifies extra Phoenix options.

jdbc:<protocol variant>[:<server list>[:<port>[:<zk root node>[:<principal>[:<keytab file>]]]]][;<option>=<value>]*

• protocol variant: The HBase connection registry to use (details below).
• server list: A comma-separated list of hostnames or IPv4 addresses. It is also possible to specify per-host ports, as defined in HBASE-12706. In this case : characters must be escaped with \. You may need to escape again in Java source strings.
• port: An integer port number. Ports specified in server list take precedence.
• zk root node: The root znode for HBase. Must be empty for non-ZK registries.
• principal: The Kerberos principal used for authentication.
  If only principal is specified, this defines a distinct user identity with its own dedicated HBase connection (HConnection) and allows multiple differently configured connections in the same JVM.
• keytab: Kerberos keytab used for authentication. Must be specified together with principal.
• option: A connection option.
• value: A connection option value.

Parameters from end of the connection definition can be omitted. Use empty strings for missing parameters in the middle of the URL. For example, the jdbc:phoenix::::principal:/home/user/keytab URL can be used to specify the kerberos principal and keytab, while using the default connection specified in hbase-site.xml.

Default connection

The underlying HBase client identifies the cluster based on parameters in hbase-site.xml. While Phoenix allows overriding this, it is usually best to use the cluster definition from hbase-site.xml. The only time the connection should be directly specified is when switching between otherwise identically configured HBase instances, like a production and a disaster recovery cluster.

To use the defaults from hbase-site.xml, use the jdbc:phoenix URL or jdbc:phoenix;option=value if additional options are needed.

See HBase documentation for how each registry is configured in hbase-site.xml.

The jdbc:phoenix: protocol variant

If this protocol variant is specified, Phoenix will select the registry based on the value of hbase.client.registry.impl.

If hbase.client.registry.impl is not defined, Phoenix chooses a default based on the HBase client version it includes.

The jdbc:phoenix+zk: protocol variant

This uses the original ZooKeeper-based HBase connection registry. The server list and port specify the ZK quorum. HBASE-12706 is supported; : characters must be escaped with \.

Examples:

• jdbc:phoenix+zk:localhost:2181:/hbase:principal:keytab - fully specified
• jdbc:phoenix+zk:host1\:2181,host1\:2182,host2\:2183 - heterogeneous ports, default ZK root node
• jdbc:phoenix+zk - use default ZK parameters from hbase-site.xml (using jdbc:phoenix is preferred in most cases)

The jdbc:phoenix+master: protocol variant

This uses the Master based connection registry added in HBASE-18095, and is available from HBase 2.3.0. The zk root node parameter must never be specified.

Examples:

• jdbc:phoenix+master:master1\:16001,master2\:16002::principal:/path/to/keytab - fully specified
• jdbc:phoenix+master:master1,master2 - use default master port for both hosts

The jdbc:phoenix+rpc: protocol variant

This uses the Master based connection registry added in HBASE-26150, and is available from HBase 2.5.0. This is very similar to the phoenix+master variant, but also allows specifying RegionServers in the host list. There is no built-in default port for this registry, the port must always be specified together with the host list.

Examples:

• jdbc:phoenix+rpc:server1\:16001,server2\:16002::principal:/path/to/keytab - fully specified
• jdbc:phoenix+rpc - use values from hbase-site.xml

Notes

Earlier versions support only the jdbc:phoenix: protocol variant implementing the original HBase ZooKeeper connection registry.

Support for registry variants is only available for HBase versions that support them. Phoenix will throw an error if a variant that the HBase client version doesn't support is specified.

Phoenix 5.2 also supports High Availability connections. Documentation for that is only available in the JIRA ticket.

Tuning Guide

Tuning Phoenix can be complex, but with a little knowledge of how it works you can make significant improvements to read and write performance. The most important factor is schema design, especially how it affects underlying HBase row keys. See "General Tips" below for design guidance based on anticipated data access patterns. Subsequent sections describe how to use secondary indexes, hints, and explain plans.

Note: Phoenix and HBase work well when your application does point lookups and small range scans. This can be achieved by good primary key design (see below). If you find that your application requires many full table scans, then Phoenix and HBase are likely not the best tool for the job. Instead, look at using other tools that write to HDFS directly using columnar representations such as Parquet.

Primary Keys

The underlying row key design is the single most important factor in Phoenix performance, and it's important to get it right at design time because you cannot change it later without re-writing the data and index tables.

The Phoenix primary keys are concatenated to create the underlying row key in Apache HBase. Choose and order primary key columns to align with common query patterns. The leading key column has the greatest performance impact. For example, if you lead with a column containing org IDs, it is easy to select all rows for a specific org. You can also add the HBase row timestamp to the primary key to improve scan efficiency by skipping rows outside the queried time range.

Every primary key imposes a cost because the entire row key is appended to every piece of data in memory and on disk. The larger the row key, the greater the storage overhead. Find ways to store information compactly in columns you plan to use for primary keys — store deltas instead of complete time stamps, for example.

To sum up, the best practice is to design primary keys to add up to a row key that lets you scan the smallest amount of data.

Tip: When choosing primary keys, lead with the column you filter most frequently across the queries that are most important to optimize. If you use ORDER BY, make sure your PK columns match expressions in the ORDER BY clause.

Monotonically increasing Primary keys

If your primary keys are monotonically increasing, use salting to help distribute writes across the cluster and improve parallelization. Example:

CREATE TABLE … ( … ) SALT_BUCKETS = N

For optimal performance the number of salt buckets should approximately equal the number of region servers. Do not salt automatically. Use salting only when experiencing hotspotting. The downside of salting is that it imposes a cost on read because when you want to query the data you have to run multiple queries to do a range scan.

General Tips

The following sections provide a few general tips for different access scenarios.

Is the Data Random-Access?

• As with any random read workloads, SSDs can improve performance because of their faster random seek time.

Is the data read-heavy or write-heavy?

• For read-heavy data:
  • Create global indexes. This will affect write speed depending on the number of columns included in an index because each index writes to its own separate table.
  • Use multiple indexes to provide fast access to common queries.
  • When specifying machines for HBase, do not skimp on cores; HBase needs them.
• For write-heavy data:
  • Pre-split the table. It can be helpful to split the table into predefined regions, or if keys are monotonically increasing, use salting to avoid creating write hotspots on a small number of nodes. Use real data types rather than raw byte data.
  • Create local indexes. Reads from local indexes have a performance penalty, so it's important to do performance testing. See the Pherf tool.

Which columns will be accessed often?

• Choose commonly-queried columns as primary keys. For more information, see “Primary Keys” below.
  • Create additional indexes to support common query patterns, including heavily accessed fields that are not in the primary key.

Can the data be append-only (immutable)?

• If the data is immutable or append-only, declare the table and its indexes as immutable using the IMMUTABLE_ROWS option at creation time to reduce the write-time cost. If you need to make an existing table immutable, you can do so with ALTER TABLE trans.event SET IMMUTABLE_ROWS=true after creation time.
  • If speed is more important than data integrity, you can use the DISABLE_WAL option. Note: it is possible to lose data with DISABLE_WAL if a region server fails.
• Set the UPDATE_CACHE_FREQUENCY option to 15 minutes or so if your metadata doesn't change very often. This property determines how often an RPC is done to ensure you're seeing the latest schema.
• If the data is not sparse (over 50% of the cells have values), use the SINGLE_CELL_ARRAY_WITH_OFFSETS data encoding scheme introduced in Phoenix 4.10, which obtains faster performance by reducing the size of the data. For more information, see “Column Mapping and Immutable Data Encoding” on the Apache Phoenix blog.

Is the table very large?

• Use the ASYNC keyword with your CREATE INDEX call to create the index asynchronously via MapReduce job. You'll need to manually start the job; see Index Population for details.
• If the data is too large to scan the table completely, use primary keys to create an underlying composite row key that makes it easy to return a subset of the data or facilitates skip-scanning — Phoenix can jump directly to matching keys when the query includes key sets in the predicate.

Is transactionality required?

A transaction is a data operation that is atomic — that is, guaranteed to succeed completely or not at all. For example, if you need to make cross-row updates to a data table, then you should consider your data transactional.

• If you need transactionality, use the TRANSACTIONAL option. (See also Transactions)

Block Encoding

Using compression or encoding is a must. Both SNAPPY and FAST_DIFF are good all around options.

FAST_DIFF encoding is automatically enabled on all Phoenix tables by default, and almost always improves overall read latencies and throughput by allowing more data to fit into blockcache. Note: FAST_DIFF encoding can increase garbage produced during request processing.

Set encoding at table creation time. Example: CREATE TABLE … ( … ) DATA_BLOCK_ENCODING=‘FAST_DIFF’

Schema Design

Because the schema affects the way the data is written to the underlying HBase layer, Phoenix performance relies on the design of your tables, indexes, and primary keys.

Phoenix and the HBase data model

HBase stores data in tables, which in turn contain columns grouped in column families. A row in an HBase table consists of versioned cells associated with one or more columns. An HBase row is a collection of many key-value pairs in which the rowkey attribute of the keys are equal. Data in an HBase table is sorted by the rowkey, and all access is via the rowkey. Phoenix creates a relational data model on top of HBase, enforcing a PRIMARY KEY constraint whose columns are concatenated to form the row key for the underlying HBase table. For this reason, it's important to be cognizant of the size and number of the columns you include in the PK constraint, because a copy of the row key is included with every cell in the underlying HBase table.

Column Families

If some columns are accessed more frequently than others, create multiple column families to separate the frequently-accessed columns from rarely-accessed columns. This improves performance because HBase reads only the column families specified in the query.

Columns

Here are a few tips that apply to columns in general, whether they are indexed or not:

• Keep VARCHAR columns under 1MB or so due to I/O costs. When processing queries, HBase materializes cells in full before sending them over to the client, and the client receives them in full before handing them off to the application code.
• For structured objects, don't use JSON, which is not very compact. Use a format such as protobuf, Avro, msgpack, or BSON.
• Consider compressing data before storage using a fast LZ variant to cut latency and I/O costs.
• Use the column mapping feature (added in Phoenix 4.10), which uses numerical HBase column qualifiers for non-PK columns instead of directly using column names. This improves performance when looking for a cell in the sorted list of cells returned by HBase, adds further across-the-board performance by reducing the disk size used by tables, and speeds up DDL operations like column rename and metadata-level column drops. For more information, see “Column Mapping and Immutable Data Encoding” on the Apache Phoenix blog.

Indexes

A Phoenix index is a physical table that stores a pivoted copy of some or all of the data in the main table to serve specific query patterns. When you issue a query, Phoenix automatically selects the best index. The primary index is created automatically based on selected primary keys. You can create secondary indexes by specifying included columns based on expected query patterns.

See also: Secondary Indexing

Secondary indexes

Secondary indexes can improve read performance by turning what would normally be a full table scan into a point lookup (at the cost of storage space and write speed). Secondary indexes can be added or removed after table creation and don't require changes to existing queries – queries simply run faster. A small number of secondary indexes is often sufficient. Depending on your needs, consider creating covered indexes or functional indexes, or both.

If your table is large, use ASYNC with CREATE INDEX to create indexes asynchronously. In this case, index build runs through MapReduce, so client restarts will not impact index creation and jobs can be retried automatically if needed. You still need to start the job manually, and then monitor it like any other MapReduce job.

Example:

CREATE INDEX IF NOT EXISTS event_object_id_idx_b
ON trans.event (object_id)
ASYNC UPDATE_CACHE_FREQUENCY = 60000;

See Index Population for details.

If you cannot create the index asynchronously, increase query timeout (phoenix.query.timeoutMs) to exceed expected index build time. If CREATE INDEX times out or the client goes down before completion, the build stops and must be run again. You can monitor the index table during creation: new regions appear as splits occur. You can query SYSTEM.STATS (populated by splits/compactions), or run COUNT(*) against the index table (higher load because it requires a full scan).

Tips:

• Create local indexes for write-heavy use cases.

• Create global indexes for read-heavy use cases. To save read-time overhead, consider creating covered indexes.

• If the primary key is monotonically increasing, create salt buckets. The salt buckets can't be changed later, so design them to handle future growth. Salt buckets help avoid write hotspots, but can decrease overall throughput due to the additional scans needed on read.

• Set up a cron job to build indexes. Use ASYNC with CREATE INDEX to avoid blocking.

• Only create the indexes you need.

• Limit the number of indexes on frequently updated tables.

• Use covered indexes to convert table scans into efficient point lookups or range queries over the index table instead of the primary table:

  CREATE INDEX idx ON table_name ( ... ) INCLUDE ( ... );

Queries

It's important to know which queries execute on the server side versus client side, because this affects performance due to network I/O and other bottlenecks. If you're querying a billion-row table, you want as much computation as possible on the server side instead of transmitting rows to the client. Some queries must still execute on the client. Sorting data that resides on multiple region servers, for example, requires aggregation and re-sort on the client.

Reading

• Avoid joins unless one side is small, especially on frequent queries. For larger joins, see “Hints,” below.
• In the WHERE clause, filter leading columns in the primary key constraint.
• Filtering the first leading column with IN or OR in the WHERE clause enables skip scan optimizations.
• Equality or comparisons (< or >) in the WHERE clause enable range-scan optimizations.
• Let Phoenix optimize query parallelism using statistics. This provides an automatic benefit if using Phoenix 4.2 or greater in production.

See also: Joins

Range Queries

If you regularly scan large data sets from spinning disk, you're best off with GZIP (but watch write speed). Use a lot of cores for a scan to utilize the available memory bandwidth. Apache Phoenix makes it easy to utilize many cores to increase scan performance.

For range queries, the HBase block cache does not provide much advantage.

Large Range Queries

For large range queries, consider setting Scan.setCacheBlocks(false) even if the whole scan could fit into the block cache.

If you mostly perform large range queries you might even want to consider running HBase with a much smaller heap and size the block cache down, to only rely on the OS Cache. This will alleviate some garbage collection related issues.

Point Lookups

For point lookups it is quite important to have your data set cached, and you should use the HBase block cache.

Hints

Hints let you override default query processing behavior and specify such factors as which index to use, what type of scan to perform, and what type of join to use.

• During the query, Hint global index if you want to force it when query includes a column not in the index.
• If necessary, you can do bigger joins with the /*+ USE_SORT_MERGE_JOIN */ hint, but a big join will be an expensive operation over huge numbers of rows.
• If the overall size of all right-hand-side tables would exceed the memory size limit, use the /*+ NO_STAR_JOIN */ hint.

See also: Hint.

Explain plans

An EXPLAIN plan tells you a lot about how a query will be run. To generate an explain plan run this query and to interpret the plan, see this reference.

Parallelization

You can improve parallelization with the UPDATE STATISTICS command. This command subdivides each region by determining keys called guideposts that are equidistant from each other, then uses these guideposts to break up queries into multiple parallel scans. Statistics are turned on by default. With Phoenix 4.9, the user can set guidepost width for each table. Optimal guidepost width depends on a number of factors such as cluster size, cluster usage, number of cores per node, table size, and disk I/O.

In Phoenix 4.12, configuration phoenix.use.stats.parallelization was added to control whether statistics are used to drive parallelization. Stats collection can still run regardless. Collected information is also used to estimate bytes and rows scanned when generating EXPLAIN.

Writing

Updating data with UPSERT VALUES

When using UPSERT VALUES to write a large number of records, turn off autocommit and batch records in reasonably small batches (try 100 rows and adjust from there to fine-tune performance).

Note: With the default fat driver, executeBatch() does not provide benefit. Instead, update multiple rows by executing UPSERT VALUES multiple times and then use commit() to submit the batch. With the thin driver, however, use executeBatch() to minimize RPCs between the client and query server.

try (Connection conn = DriverManager.getConnection(url)) {
  conn.setAutoCommit(false);
  int batchSize = 0;
  int commitSize = 1000; // number of rows you want to commit per batch.
  try (PreparedStatement stmt = conn.prepareStatement(upsert)) {
    // set params...
    while (/* there are records to upsert */) {
      stmt.executeUpdate();
      batchSize++;
      if (batchSize % commitSize == 0) {
        conn.commit();
      }
   }
 conn.commit(); // commit the last batch of records
 }

Note: Because the Phoenix client keeps uncommitted rows in memory, be careful not to set commitSize too high.

Updating data with UPSERT SELECT

When using UPSERT SELECT to write many rows in a single statement, turn on autocommit and the rows will be automatically batched according to the phoenix.mutate.batchSize. This will minimize the amount of data returned back to the client and is the most efficient means of updating many rows.

Deleting data

When deleting a large data set, turn on autoCommit before issuing the DELETE query so that the client does not need to remember the row keys of all the keys as they are deleted. This prevents the client from buffering the rows affected by the DELETE so that Phoenix can delete them directly on the region servers without the expense of returning them to the client.

Reducing RPC traffic

To reduce RPC traffic, set UPDATE_CACHE_FREQUENCY (4.7 or above) on your tables and indexes when creating them (or via ALTER TABLE/ALTER INDEX). See Altering.

Using local indexes

If using 4.8, consider using local indexes to minimize the write time. In this case, the writes for the secondary index will be to the same region server as your base table. This approach does involve a performance hit on the read side, though, so make sure to quantify both write speed improvement and read speed reduction.

Further tuning

For advice about tuning the underlying HBase and JVM layers, see Operational and Performance Configuration Options in the Apache HBase™ Reference Guide.

Special Cases

The following sections provide Phoenix-specific additions to the tuning recommendations in the Apache HBase™ Reference Guide section referenced above.

For applications where failing quickly is better than waiting

In addition to the HBase tuning referenced above, set phoenix.query.timeoutMs in hbase-site.xml on the client side to the maximum tolerable wait time in milliseconds.

For applications that can tolerate slightly out of date information

In addition to the HBase tuning referenced above, set phoenix.connection.consistency = timeline in hbase-site.xml on the client side for all connections.

Installation

Installation

To install a pre-built Phoenix, use these directions:

• Download and expand the latest phoenix-hbase-[hbase.version]-[phoenix.version]-bin.tar.gz for your HBase version.
• Add phoenix-server-hbase-[hbase.version]-[phoenix.version].jar to the classpath of all HBase region servers and masters, and remove any previous version. An easy way is to copy it into the HBase lib directory.
• Restart HBase.
• Add phoenix-client-hbase-[hbase.version]-[phoenix.version].jar to the classpath of any JDBC client.

To install Phoenix from source:

• Download and expand the latest phoenix-[phoenix.version]-src.tar.gz for your HBase version, or check it out from the main source repository.
• Follow the build instructions in BUILDING.md in the root directory of the source distribution/repository to build the binary assembly.
• Follow the instructions above, but use the assembly built from source.

Getting Started

Want to get started quickly? Take a look at our FAQs and quick start guide here.

Command Line

A terminal interface to execute SQL from the command line is now bundled with Phoenix. To start it, execute the following from the bin directory:

$ sqlline.py [zk quorum hosts]

To execute SQL scripts from the command line, you can include a SQL file argument like this:

$ sqlline.py [zk quorum hosts] ../examples/stock_symbol.sql

For more information, see the manual.

Loading Data

In addition, you can use bin/psql.py to load CSV data or execute SQL scripts. For example:

$ psql.py localhost ../examples/web_stat.sql ../examples/web_stat.csv ../examples/web_stat_queries.sql

Other alternatives include:

• Using our map-reduce based CSV loader for bigger data sets
• Mapping an existing HBase table to a Phoenix table and using the UPSERT SELECT command to populate a new table.
• Populating the table through our UPSERT VALUES command.

SQuirreL SQL Client

If you'd rather use a client GUI to interact with Phoenix, download and install SQuirrel. Since Phoenix is a JDBC driver, integration with tools such as this are seamless. Here are the setup steps necessary:

Through SQuirreL, you can issue SQL statements in the SQL tab (create tables, insert data, run queries), and inspect table metadata in the Object tab (for example, list tables, columns, primary keys, and types).

Note that most graphical clients that support generic JDBC drives should also work, and the setup process is usually similar.

Samples

The best place to see samples are in our unit tests under src/test/java. The ones in the endToEnd package are tests demonstrating how to use all aspects of the Phoenix JDBC driver. We also have some examples in the examples directory.

Configuration

Configuration

Phoenix provides many different knobs and dials to configure and tune the system to run more optimally on your cluster. The configuration is done through a series of Phoenix-specific properties specified both on client and server-side hbase-site.xml files. In addition to these properties, there are of course all the HBase configuration properties with the most important ones documented here.

The table below outlines the full set of Phoenix-specific configuration properties and their defaults.

Backward Compatibility

Phoenix maintains backward compatibility across at least two minor releases to allow for no downtime through server-side rolling restarts upon upgrading. See below for details.

Versioning Convention

Phoenix uses a standard three number versioning schema of the form:

<major version> . <minor version> . <patch version>

For example, 4.2.1 has a major version of 4, a minor version of 2, and a patch version of 1.

Patch Release

Upgrading to a new patch release (i.e. only the patch version has changed) is the simplest case. The jar upgrade may occur in any order: client first or server first, and a mix of clients with different patch release versions is fine.

Minor Release

When upgrading to a new minor release (i.e. the major version is the same, but the minor version has changed), sometimes modifications to the system tables are necessary to either fix a bug or provide a new feature. This upgrade will occur automatically the first time a newly upgraded client connects to the newly upgraded server. It is required that the server-side jar be upgraded first across your entire cluster, before any clients are upgraded. An older client (two minor versions back) will work with a newer server jar when the minor version is different, but not vice versa. In other words, clients do not need to be upgraded in lock step with the server. However, as the server version moves forward, the client version should move forward as well. This allows Phoenix to evolve its client/server protocol while still providing clients sufficient time to upgrade their clients.

As of the 4.3 release, a mix of clients on different minor release versions is supported as well (note that prior releases required all clients to be upgraded at the same time). Another improvement as of the 4.3 release is that an upgrade may be done directly from one minor version to another higher minor version (prior releases required an upgrade to each minor version in between).

Major Release

Upgrading to a new major release may require downtime as well as potentially the running of a migration script. Additionally, all clients and servers may need to be upgraded at the same time. This will be determined on a release-by-release basis.

Release Notes

Specific details on issues and their fixes that may impact you can be found here.

Performance

Phoenix follows the philosophy of bringing the computation to the data by using:

• coprocessors to perform operations on the server-side thus minimizing client/server data transfer
• custom filters to prune data as close to the source as possible.

In addition, to minimize startup costs, Phoenix uses native HBase APIs rather than going through the MapReduce framework.

Phoenix vs related products

Below are charts showing relative performance between Phoenix and some other related products.

Phoenix vs Hive (running over HDFS and HBase)

Query: SELECT COUNT(1) from a table over 10M and 100M rows. Data has 5 narrow columns. Number of region servers: 4 (HBase heap: 10GB, processor: 6 cores @ 3.3GHz Xeon).

Phoenix vs Impala (running over HBase)

Query: SELECT COUNT(1) from a table over 1M and 5M rows. Data has 3 narrow columns. Number of region servers: 1 (virtual machine, HBase heap: 2GB, processor: 2 cores @ 3.3GHz Xeon).

Latest Automated Performance Run

Latest Automated Performance Run | Automated Performance Runs History

Performance improvements in Phoenix 1.2

Essential Column Family

The Phoenix 1.2 query filter leverages the HBase Filter Essential Column Family feature. This improves performance when Phoenix filters data split across multiple column families (CFs) by loading only essential CFs first. In a second pass, all CFs are loaded as needed.

Consider the following schema in which data is split into two CFs: CREATE TABLE t (k VARCHAR NOT NULL PRIMARY KEY, a.c1 INTEGER, b.c2 VARCHAR, b.c3 VARCHAR, b.c4 VARCHAR).

Running a query similar to the following shows significant performance gains when a subset of rows matches the filter: SELECT COUNT(c2) FROM t WHERE c1 = ?

The following chart shows in-memory query performance for the query above with 10M rows on 4 region servers, when 10% of rows match the filter. Note: cf-a is approximately 8 bytes and cf-b is approximately 400 bytes wide.

Skip Scan

Skip Scan Filter leverages HBase filter SEEK_NEXT_USING_HINT (docs). It significantly improves point queries over key columns.

Consider the following schema in which data is split into two CFs: CREATE TABLE t (k VARCHAR NOT NULL PRIMARY KEY, a.c1 INTEGER, b.c2 VARCHAR, b.c3 VARCHAR).

Running a query similar to the following shows significant performance gains when a subset of rows matches the filter: SELECT COUNT(c1) FROM t WHERE k IN (1% random k's)

The following chart shows in-memory query performance of the query above with 10M rows on 4 region servers when 1% random keys over the full key range are passed in the IN clause. Note: all VARCHAR columns are approximately 15 bytes.

Salting

Salting in Phoenix 1.2 improves both read and write performance by adding an extra hash byte at the start of the key and pre-splitting data into regions. This reduces hotspotting on one or a few region servers. Read more about this feature here.

Consider the following schema:

CREATE TABLE T (HOST CHAR(2) NOT NULL,DOMAIN VARCHAR NOT NULL, FEATURE VARCHAR NOT NULL,DATE DATE NOT NULL,USAGE.CORE BIGINT,USAGE.DB BIGINT,STATS.ACTIVE_VISITOR INTEGER CONSTRAINT PK PRIMARY KEY (HOST, DOMAIN, FEATURE, DATE)) SALT_BUCKETS = 4.

The following chart shows write performance with and without salting, where the table is split into 4 regions on a 4-region-server cluster (note: for optimal performance, salt bucket count should match region server count).

The following chart shows in-memory query performance for a 10M-row table where host='NA' matches 3.3M rows.

select count(1) from t where host='NA'

Top-N

The following chart shows in-memory query time for a Top-N query over 10M rows using Phoenix 1.2 and Hive over HBase.

select core from t order by core desc limit 10

Performance Testing

Overview

Pherf is a standalone tool for performance and functional testing through Phoenix. Pherf can be used both to generate highly customized datasets and to measure SQL performance against that data.

Build all of Phoenix (includes Pherf default profile)

mvn clean package -DskipTests

Running

• Edit config/env.sh to include the required property values.
• bin/pherf-standalone.py -h
• Example: bin/pherf-standalone.py -drop all -l -q -z [zookeeper] -schemaFile .*user_defined_schema.sql -scenarioFile .*user_defined_scenario.xml

Example run commands

List all scenario files available to run.

./pherf-standalone.py -listFiles

Drop all existing tables, load and query data specified in all scenario files.

./pherf-standalone.py -drop all -l -q -z localhost

Pherf arguments:

• -h Help
• -l Apply schema and load data
• -q Executes multi-threaded query sets and writes results
• -z [quorum] ZooKeeper quorum
• -m Enable monitor for statistics
• -monitorFrequency [frequency in ms] Frequency at which the monitor will snapshot stats to log file
• -drop [pattern] Regex drop all tables with schema name as PHERF. Example drop Event tables: -drop .*(EVENT).*. Drop all: -drop .* or -drop all
• -scenarioFile Regex or file name of a specific scenario file to run
• -schemaFile Regex or file name of a specific schema file to run
• -export Exports query results to CSV files in CSV_EXPORT directory
• -diff Compares results with previously exported results
• -hint Executes all queries with specified hint. Example: SMALL
• -rowCountOverride
• -rowCountOverride [number of rows] Specify number of rows to be upserted rather than using row count specified in schema

Adding Rules for Data Creation

Review test_scenario.xml for syntax examples.

• Rules are defined as <columns /> and are applied in the order they appear in file.
• Rules of the same type override the values of a prior rule of the same type. If <userDefined>true</userDefined> is set, rule will only apply override when type and name match the column name in Phoenix.
• <prefix> tag is set at the column level. It can be used to define a constant string appended to the beginning of CHAR and VARCHAR data type values.
• Required field Supported Phoenix types: VARCHAR, CHAR, DATE, DECIMAL, INTEGER
  • denoted by the <type> tag
• User defined true changes rule matching to use both name and type fields to determine equivalence.
  • Default is false if not specified and equivalence will be determined by type only. An important note here is that you can still override rules without the user defined flag, but they will change the rule globally and not just for a specified column.
• Required field Supported Data Sequences
  • RANDOM: Random value which can be bound by other fields such as length.
  • SEQUENTIAL: Monotonically increasing long prepended to random strings.
    • Only supported on VARCHAR and CHAR types
  • LIST: Means pick values from predefined list of values
• Required field Length defines boundary for random values for CHAR and VARCHAR types.
  • denoted by the <length> tag
• Column level Min/Max value defines boundaries for numerical values. For DATEs, these values supply a range between which values are generated. At the column level the granularity is a year. At a specific data value level, the granularity is down to the Ms.
  • denoted by the <minValue> tag
  • denoted by the <maxValue> tag
• Null chance denotes the probability of generating a null value. From [0-100]. The higher the number, the more likely the value will be null. Denoted by <nullChance>.
• Name can either be any text or the actual column name in the Phoenix table.
  • denoted by the <name> tag
• Value List is used in conjunction with LIST data sequences. Each entry is a DataValue with a specified value to be used when generating data.
  • Denoted by the <valueList><datavalue><value/></datavalue></valueList> tags
  • If the distribution attribute on the datavalue is set, values will be created according to that probability.
  • When distribution is used, values must add up to 100%.
  • If distribution is not used, values will be randomly picked from the list with equal distribution.

Defining Scenario

A scenario can have multiple querySets. Consider the following example: concurrency of 1-4 means that each query will be executed starting with concurrency level of 1 and reach up to maximum concurrency of 4. Per thread, query would be executed to a minimum of 10 times or 10 seconds (whichever comes first). QuerySet by default is executed serially but you can change executionType to PARALLEL so queries are executed concurrently. Each Query may have an optional timeoutDuration field that defines the amount of time (in milliseconds) before execution for that Query is cancelled. Scenarios are defined in XML files stored in the resource directory.

<scenarios>
  <querySet concurrency="1-4" executionType="PARALLEL" executionDurationInMs="10000" numberOfExecutions="10">
    <query id="q1" verifyRowCount="false" statement="select count(*) from PHERF.TEST_TABLE"/>
    <query id="q2" tenantId="1234567890" timeoutDuration="10000" ddl="create view if not exists myview(mypk varchar not null primary key, mycol varchar)" statement="upsert select ..."/>
  </querySet>
  <querySet concurrency="3" executionType="SERIAL" executionDurationInMs="20000" numberOfExecutions="100">
    <query id="q3" verifyRowCount="false" statement="select count(*) from PHERF.TEST_TABLE"/>
    <query id="q4" statement="select count(*) from PHERF.TEST_TABLE WHERE TENANT_ID='00D000000000062'"/>
  </querySet>
</scenarios>

Results

Results are written real time in results directory. Open the result that is saved in .jpg format for real time visualization. Results are written using DataModelResult objects, which are modified over the course of each Pherf run.

XML results

Pherf XML results have a similar format to the corresponding scenario.xml file used for the Pherf run, but also include additional information, such as the execution time of queries, whether queries timed out, and result row count.

<queryResults expectedAggregateRowCount="100000" id="q1" statement="SELECT COUNT(*) FROM PHERF.USER_DEFINED_TEST" timeoutDuration="0">
  <threadTimes threadName="1,1">
    <runTimesInMs elapsedDurationInMs="1873" resultRowCount="100000" startTime="2020-04-09T11:28:12.623-07:00" timedOut="true"/>
    <runTimesInMs elapsedDurationInMs="1793" resultRowCount="100000" startTime="2020-04-09T11:28:14.511-07:00" timedOut="true"/>
    <runTimesInMs elapsedDurationInMs="1764" resultRowCount="100000" startTime="2020-04-09T11:28:16.319-07:00" timedOut="true"/>
  </threadTimes>
</queryResults>

CSV results

Each row in a CSV result file represents a single execution of a query and provides details about a query execution's runtime, timeout status, result row count, and more. The header file format can be found in Header.java.

Testing

Default quorum is localhost. If you want to override set the system variable.

Run unit tests: mvn test -DZK_QUORUM=localhost
Run a specific method: mvn -Dtest=ClassName#methodName test

More to come...

Integrations

Spark Integration

The phoenix-spark plugin extends Phoenix's MapReduce support to allow Spark to load Phoenix tables as DataFrames, and enables persisting them back to Phoenix.

Prerequisites

• Phoenix 4.4.0+
• Spark 1.3.1+ (prebuilt with Hadoop 2.4 recommended)

Why not JDBC?

Although Spark supports connecting directly to JDBC databases, it's only able to parallelize queries by partitioning on a numeric column. It also requires a known lower bound, upper bound and partition count in order to create split queries.

In contrast, the phoenix-spark integration is able to leverage the underlying splits provided by Phoenix in order to retrieve and save data across multiple workers. All that's required is a database URL and a table name. Optional SELECT columns can be given, as well as pushdown predicates for efficient filtering.

The choice of which method to use to access Phoenix comes down to each specific use case.

Spark setup

• To ensure that all requisite Phoenix/HBase platform dependencies are available on the classpath for Spark executors and drivers, set both spark.executor.extraClassPath and spark.driver.extraClassPath in spark-defaults.conf to include phoenix-<version>-client.jar.

• Note that for Phoenix versions 4.7 and 4.8 you must use the 'phoenix-<version>-client-spark.jar'.

• As of Phoenix 4.10, phoenix-<version>-client.jar is compiled against Spark 2.x. If compatibility with Spark 1.x is needed, compile Phoenix with the spark16 Maven profile.

• To help your IDE, you can add the following provided dependency to your build:

  <dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-spark</artifactId>
    <version>${phoenix.version}</version>
    <scope>provided</scope>
  </dependency>

• As of Phoenix 4.15.0, the connectors project is separated from the main Phoenix project (see phoenix-connectors) and will have its own releases. You can add the following dependency in your project:

  <dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-spark</artifactId>
    <version>${phoenix.connectors.version}</version>
  </dependency>

The first released connectors jar is connectors-1.0.0 (replace phoenix.connectors.version above with this version).

Reading Phoenix Tables

Given a Phoenix table with the following DDL and DML:

CREATE TABLE TABLE1 (ID BIGINT NOT NULL PRIMARY KEY, COL1 VARCHAR);
UPSERT INTO TABLE1 (ID, COL1) VALUES (1, 'test_row_1');
UPSERT INTO TABLE1 (ID, COL1) VALUES (2, 'test_row_2');

Load as a DataFrame using the DataSourceV2 API

Scala example:

import org.apache.spark.SparkContext
import org.apache.spark.sql.{SQLContext, SparkSession}
import org.apache.phoenix.spark.datasource.v2.PhoenixDataSource

val spark = SparkSession
  .builder()
  .appName("phoenix-test")
  .master("local")
  .getOrCreate()

// Load data from TABLE1
val df = spark.sqlContext
  .read
  .format("phoenix")
  .options(Map("table" -> "TABLE1", PhoenixDataSource.ZOOKEEPER_URL -> "phoenix-server:2181"))
  .load

df.filter(df("COL1") === "test_row_1" && df("ID") === 1L)
  .select(df("ID"))
  .show

Java example:

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import static org.apache.phoenix.spark.datasource.v2.PhoenixDataSource.ZOOKEEPER_URL;

public class PhoenixSparkRead {

    public static void main() throws Exception {
        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("phoenix-test");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(jsc);

        // Load data from TABLE1
        Dataset<Row> df = sqlContext
            .read()
            .format("phoenix")
            .option("table", "TABLE1")
            .option(ZOOKEEPER_URL, "phoenix-server:2181")
            .load();
        df.createOrReplaceTempView("TABLE1");

        SQLContext sqlCtx = new SQLContext(jsc);
        df = sqlCtx.sql("SELECT * FROM TABLE1 WHERE COL1='test_row_1' AND ID=1L");
        df.show();
        jsc.stop();
    }
}

Saving to Phoenix

Save DataFrames to Phoenix using DataSourceV2

The save method on DataFrame allows passing in a data source type. You can use phoenix for DataSourceV2 and must also pass in a table and zkUrl parameter to specify which table and server to persist the DataFrame to. The column names are derived from the DataFrame's schema field names, and must match the Phoenix column names.

The save method also takes a SaveMode option, for which only SaveMode.Overwrite is supported.

Given two Phoenix tables with the following DDL:

CREATE TABLE INPUT_TABLE (id BIGINT NOT NULL PRIMARY KEY, col1 VARCHAR, col2 INTEGER);
CREATE TABLE OUTPUT_TABLE (id BIGINT NOT NULL PRIMARY KEY, col1 VARCHAR, col2 INTEGER);

you can load from an input table and save to an output table as a DataFrame as follows in Scala:

import org.apache.spark.SparkContext
import org.apache.spark.sql.{SQLContext, SparkSession, SaveMode}
import org.apache.phoenix.spark.datasource.v2.PhoenixDataSource

val spark = SparkSession
  .builder()
  .appName("phoenix-test")
  .master("local")
  .getOrCreate()

// Load INPUT_TABLE
val df = spark.sqlContext
  .read
  .format("phoenix")
  .options(Map("table" -> "INPUT_TABLE", PhoenixDataSource.ZOOKEEPER_URL -> "phoenix-server:2181"))
  .load

// Save to OUTPUT_TABLE
df.write
  .format("phoenix")
  .mode(SaveMode.Overwrite)
  .options(Map("table" -> "OUTPUT_TABLE", PhoenixDataSource.ZOOKEEPER_URL -> "phoenix-server:2181"))
  .save()

Java example:

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SQLContext;

import static org.apache.phoenix.spark.datasource.v2.PhoenixDataSource.ZOOKEEPER_URL;

public class PhoenixSparkWriteFromInputTable {

    public static void main() throws Exception {
        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("phoenix-test");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(jsc);

        // Load INPUT_TABLE
        Dataset<Row> df = sqlContext
            .read()
            .format("phoenix")
            .option("table", "INPUT_TABLE")
            .option(ZOOKEEPER_URL, "phoenix-server:2181")
            .load();

        // Save to OUTPUT_TABLE
        df.write()
          .format("phoenix")
          .mode(SaveMode.Overwrite)
          .option("table", "OUTPUT_TABLE")
          .option(ZOOKEEPER_URL, "phoenix-server:2181")
          .save();
        jsc.stop();
    }
}

Save from an external RDD with a schema to a Phoenix table

Just like the previous example, you can pass in the data source type as phoenix and specify the table and zkUrl parameters indicating which table and server to persist the DataFrame to.

Note that the schema of the RDD must match its column data and this must match the schema of the Phoenix table that you save to.

Given an output Phoenix table with the following DDL:

CREATE TABLE OUTPUT_TABLE (id BIGINT NOT NULL PRIMARY KEY, col1 VARCHAR, col2 INTEGER);

you can save a dataframe from an RDD as follows in Scala:

import org.apache.spark.SparkContext
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType, StructField}
import org.apache.spark.sql.{Row, SQLContext, SparkSession, SaveMode}
import org.apache.phoenix.spark.datasource.v2.PhoenixDataSource

val spark = SparkSession
  .builder()
  .appName("phoenix-test")
  .master("local")
  .getOrCreate()

val dataSet = List(Row(1L, "1", 1), Row(2L, "2", 2), Row(3L, "3", 3))

val schema = StructType(
  Seq(StructField("ID", LongType, nullable = false),
    StructField("COL1", StringType),
    StructField("COL2", IntegerType)))

val rowRDD = spark.sparkContext.parallelize(dataSet)

// Apply the schema to the RDD.
val df = spark.sqlContext.createDataFrame(rowRDD, schema)

df.write
  .format("phoenix")
  .options(Map("table" -> "OUTPUT_TABLE", PhoenixDataSource.ZOOKEEPER_URL -> "phoenix-server:2181"))
  .mode(SaveMode.Overwrite)
  .save()

Java example:

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;

import static org.apache.phoenix.spark.datasource.v2.PhoenixDataSource.ZOOKEEPER_URL;

public class PhoenixSparkWriteFromRDDWithSchema {

    public static void main() throws Exception {
        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("phoenix-test");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(jsc);
        SparkSession spark = sqlContext.sparkSession();
        Dataset<Row> df;

        // Generate the schema based on the fields
        List<StructField> fields = new ArrayList<>();
        fields.add(DataTypes.createStructField("ID", DataTypes.LongType, false));
        fields.add(DataTypes.createStructField("COL1", DataTypes.StringType, true));
        fields.add(DataTypes.createStructField("COL2", DataTypes.IntegerType, true));
        StructType schema = DataTypes.createStructType(fields);

        // Generate the rows with the same exact schema
        List<Row> rows = new ArrayList<>();
        for (int i = 1; i < 4; i++) {
            rows.add(RowFactory.create(Long.valueOf(i), String.valueOf(i), i));
        }

        // Create a DataFrame from the rows and the specified schema
        df = spark.createDataFrame(rows, schema);
        df.write()
            .format("phoenix")
            .mode(SaveMode.Overwrite)
            .option("table", "OUTPUT_TABLE")
            .option(ZOOKEEPER_URL,  "phoenix-server:2181")
            .save();

        jsc.stop();
    }
}

PySpark

With Spark's DataFrame support, you can also use pyspark to read and write from Phoenix tables.

Load a DataFrame

Given a table TABLE1 and a ZooKeeper URL of phoenix-server:2181, you can load the table as a DataFrame using the following Python code in pyspark:

df = sqlContext.read \
  .format("phoenix") \
  .option("table", "TABLE1") \
  .option("zkUrl", "phoenix-server:2181") \
  .load()

Save a DataFrame

Given the same table and ZooKeeper URL above, you can save a DataFrame to a Phoenix table using the following code:

df.write \
  .format("phoenix") \
  .mode("overwrite") \
  .option("table", "TABLE1") \
  .option("zkUrl", "phoenix-server:2181") \
  .save()

Notes

• If you want to use DataSourceV1, you can use source type "org.apache.phoenix.spark" instead of "phoenix", however this is deprecated as of connectors-1.0.0.

• The (deprecated) functions phoenixTableAsDataFrame, phoenixTableAsRDD and saveToPhoenix all support optionally specifying a conf Hadoop configuration parameter with custom Phoenix client settings, as well as an optional zkUrl parameter for the Phoenix connection URL.

• If zkUrl isn't specified, it's assumed that the "hbase.zookeeper.quorum" property has been set in the conf parameter. Similarly, if no configuration is passed in, zkUrl must be specified.

• As of PHOENIX-5197, you can pass configurations from the driver to executors as a comma-separated list against the key phoenixConfigs i.e (PhoenixDataSource.PHOENIX_CONFIGS), for ex:

  df = spark
    .sqlContext
    .read
    .format("phoenix")
    .options(Map("table" -> "Table1", "zkUrl" -> "phoenix-server:2181",
      "phoenixConfigs" -> "hbase.client.retries.number=10,hbase.client.pause=10000"))
    .load;

This list of properties is parsed and populated into a properties map which is passed to DriverManager.getConnection(connString, propsMap). Note that the same property values will be used for both the driver and all executors and these configurations are used each time a connection is made (both on the driver and executors).

Limitations

• Basic support for column and predicate pushdown using the Data Source API
• The Data Source API does not support passing custom Phoenix settings in configuration, you must create the DataFrame or RDD directly if you need fine-grained configuration.
• No support for aggregate or distinct queries as explained in our Map Reduce Integration documentation.

PageRank example

This example makes use of the Enron email data set, provided by the Stanford Network Analysis Project, and executes the GraphX implementation of PageRank on it to find interesting entities. It then saves the results back to Phoenix.

1. Download and extract the file enron.csv.gz

2. Create the necessary Phoenix schema

   CREATE TABLE EMAIL_ENRON(MAIL_FROM BIGINT NOT NULL, MAIL_TO BIGINT NOT NULL CONSTRAINT pk PRIMARY KEY(MAIL_FROM, MAIL_TO));
   CREATE TABLE EMAIL_ENRON_PAGERANK(ID BIGINT NOT NULL, RANK DOUBLE CONSTRAINT pk PRIMARY KEY(ID));

3. Load the email data into Phoenix (assuming localhost for ZooKeeper quorum URL)

   gunzip /tmp/enron.csv.gz
   cd /path/to/phoenix/bin
   ./psql.py -t EMAIL_ENRON localhost /tmp/enron.csv

4. In spark-shell, with the phoenix-client in the Spark driver classpath, run the following:

   import org.apache.spark.graphx._
   import org.apache.phoenix.spark._
   val rdd = sc.phoenixTableAsRDD("EMAIL_ENRON", Seq("MAIL_FROM", "MAIL_TO"), zkUrl=Some("localhost"))           // load from phoenix
   val rawEdges = rdd.map{ e => (e("MAIL_FROM").asInstanceOf[VertexId], e("MAIL_TO").asInstanceOf[VertexId]) }   // map to vertexids
   val graph = Graph.fromEdgeTuples(rawEdges, 1.0)                                                               // create a graph
   val pr = graph.pageRank(0.001)                                                                                // run pagerank
   pr.vertices.saveToPhoenix("EMAIL_ENRON_PAGERANK", Seq("ID", "RANK"), zkUrl = Some("localhost"))               // save to phoenix

5. Query the top ranked entities in SQL

   SELECT * FROM EMAIL_ENRON_PAGERANK ORDER BY RANK DESC LIMIT 5;
   +------------------------------------------+------------------------------------------+
   |                    ID                    |                   RANK                   |
   +------------------------------------------+------------------------------------------+
   | 5038                                     | 497.2989872977676                        |
   | 273                                      | 117.18141799210386                       |
   | 140                                      | 108.63091596789913                       |
   | 458                                      | 107.2728800448782                        |
   | 588                                      | 106.11840798585399                       |
   +------------------------------------------+------------------------------------------+

Deprecated Usages

Load as a DataFrame directly using a Configuration object

import org.apache.hadoop.conf.Configuration
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.phoenix.spark._

val configuration = new Configuration()
// Can set Phoenix-specific settings, requires 'hbase.zookeeper.quorum'

val sc = new SparkContext("local", "phoenix-test")
val sqlContext = new SQLContext(sc)

// Load the columns 'ID' and 'COL1' from TABLE1 as a DataFrame
val df = sqlContext.phoenixTableAsDataFrame(
  "TABLE1", Array("ID", "COL1"), conf = configuration
)

df.show

Load as an RDD, using a Zookeeper URL

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.phoenix.spark._
import org.apache.spark.rdd.RDD

val sc = new SparkContext("local", "phoenix-test")

// Load the columns 'ID' and 'COL1' from TABLE1 as an RDD
val rdd: RDD[Map[String, AnyRef]] = sc.phoenixTableAsRDD(
  "TABLE1", Seq("ID", "COL1"), zkUrl = Some("phoenix-server:2181")
)

rdd.count()

val firstId = rdd.first()("ID").asInstanceOf[Long]
val firstCol = rdd.first()("COL1").asInstanceOf[String]

Saving RDDs to Phoenix

saveToPhoenix is an implicit method on RDD[Product], or an RDD of Tuples. The data types must correspond to the Java types Phoenix supports Datatypes

Given a Phoenix table with the following DDL:

CREATE TABLE OUTPUT_TEST_TABLE (id BIGINT NOT NULL PRIMARY KEY, col1 VARCHAR, col2 INTEGER);

import org.apache.spark.SparkContext
import org.apache.phoenix.spark._

val sc = new SparkContext("local", "phoenix-test")
val dataSet = List((1L, "1", 1), (2L, "2", 2), (3L, "3", 3))

sc
  .parallelize(dataSet)
  .saveToPhoenix(
    "OUTPUT_TEST_TABLE",
    Seq("ID","COL1","COL2"),
    zkUrl = Some("phoenix-server:2181")
  )

Apache Hive

The Apache Phoenix Storage Handler is a plugin that enables Apache Hive access to Phoenix tables from the Apache Hive command line using HiveQL.

Prerequisites

This document describes the plugin available in the phoenix-connectors source repo, as of December 2023.

• Phoenix 5.1.0+
• Hive 3.1.2+
• phoenix-connectors 6.0.0-SNAPSHOT

The Phoenix Storage handler currently only supports Hive 3.1. It has only been tested with Hive Hive 3.1.3, and Phoenix 5.1.3.

A variant for Hive 4 is planned after Hive 4.0.0 is released.

Building

The Phoenix Storage Handler used to be part of the main Phoenix repo, but it has been refactored into the separate phoenix-connectors repo after the release of Phoenix 5.0. At the time of writing there is no released version of the connectors project, so it must be built from the Git source repository HEAD.

Official releases will be available on the Downloads page.

Check the value of the hbase.version property in the root pom.xml. If it is older than HBase 2.5.0, then you need to rebuild HBase locally as described in BUILDING.md in the main Phoenix repository.

Check out the HEAD version of https://github.com/apache/phoenix-connectors. Build it with:

mvn clean package

The binary distribution will be created in phoenix5-connectors-assembly/target.

The driver is built for HBase 2.4.x by default.

To build it for other HBase versions, set hbase.version, hbase.compat.version, hadoop.version, zookeeper.version, and hbase-thirdparty-version to the versions used by HBase. hbase.compat.version is the matching hbase-compat module in Phoenix; other versions can be copied from the root pom.xml in the HBase source.

For example, to build with HBase 2.1.10 (rebuilt with -Dhadoop.profile=3.0), use:

mvn clean package -Dhbase.version=2.1.10 -Dhbase.compat.version=2.1.6 -Dhadoop.version=3.0.3 -Dzookeeper.version=3.4.10 -Dhbase-thirdparty-version=2.1.0

Preparing Hive 3

Hive 3.1 ships with HBase 2.0 beta, which is incompatible with Phoenix. To use the Phoenix Storage handler, HBase and its dependencies have to be removed from Hive.

To remove the shipped HBase 2.0 beta perform the following on each Hive node:

1. Create a backup of the Hive /lib directory

2. Remove the HBase dependencies from the /lib directory:

   mkdir ../lib-removed
   mv hbase-* javax.inject* jcodings-* jersey-* joni-* osgi-resource-locator-* ../lib-removed/

Even though Hive ships with HBase jars, it includes a mechanism to load the hbase mapredcp jars automatically.

To ensure that Hive finds and uses the correct HBase JARs and hbase-site.xml, set the following environment variables in hive-env.sh on each node, or make sure they are set properly in the environment:

• HBASE_HOME: The root directory of the HBase installation.
• HBASE_CONF_DIR: The directory where hbase-site.xml resides. Defaults to /etc/hbase/conf, or if it does not exist, to $HBASE_HOME/conf.
• HBASE_BIN_DIR: The directory that holds the hbase command. Defaults to $HBASE_HOME/bin.

It is assumed that the Hadoop variables are already set correctly, and the HBase libraries and up-to-date hbase-site.xml are available on each Hive node.

Hive Setup

It is necessary to make the phoenix5-hive-shaded-6.0.0-SNAPSHOT-shaded.jar available for every hive component. There are many ways to achieve this; one of the simpler options is to use the HIVE_AUX_JARS_PATH environment variable.

If hive-env.sh already sets HIVE_AUX_JARS_PATH, then copy the connector JAR there. Otherwise, create a world-readable directory on the system and copy the connector JAR there. Then add:

HIVE_AUX_JARS_PATH=<PATH TO DIRECTORY>

to hive-env.sh.

This must be performed on every Hive node.

Table Creation and Deletion

The Phoenix Storage Handler supports only EXTERNAL Hive tables.

Create EXTERNAL Table

For EXTERNAL tables, Hive works with an existing Phoenix table and manages only Hive metadata. Dropping an EXTERNAL table from Hive deletes only Hive metadata but does not delete the Phoenix table.

create external table ext_table (
  i1 int,
  s1 string,
  f1 float,
  d1 decimal
)
STORED BY 'org.apache.phoenix.hive.PhoenixStorageHandler'
TBLPROPERTIES (
  "phoenix.table.name" = "ext_table",
  "phoenix.zookeeper.quorum" = "localhost",
  "phoenix.zookeeper.znode.parent" = "/hbase",
  "phoenix.zookeeper.client.port" = "2181",
  "phoenix.rowkeys" = "i1",
  "phoenix.column.mapping" = "i1:i1, s1:s1, f1:f1, d1:d1"
);

Properties

1. phoenix.table.name
   • Specifies the Phoenix table name
   • Default: the same as the Hive table
2. phoenix.zookeeper.quorum
   • Specifies the ZooKeeper quorum for HBase
   • Default: localhost
3. phoenix.zookeeper.znode.parent
   • Specifies the ZooKeeper parent node for HBase
   • Default: /hbase
4. phoenix.zookeeper.client.port
   • Specifies the ZooKeeper port
   • Default: 2181
5. phoenix.rowkeys
   • The list of columns to be the primary key in a Phoenix table
   • Required
6. phoenix.column.mapping
   • Mappings between column names for Hive and Phoenix. See Limitations for details.

The phoenix.zookeeper.* properties are optional. If they are not specified, values from hbase-site.xml will be used.

Data Ingestion, Deletions, and Updates

Data ingestion can be done by all ways that Hive and Phoenix support:

Hive:

insert into table T values (....);
insert into table T select c1,c2,c3 from source_table;

Phoenix:

upsert into table T values (.....);

Phoenix CSV BulkLoad tools can also be used.

All delete and update operations should be performed on the Phoenix side. See Limitations for more details.

Additional Configuration Options

Those options can be set in a Hive command-line interface (CLI) environment.

Performance Tuning

Disabling WAL can lead to data loss.

Query Data

You can use HiveQL for querying data in a Phoenix table. A Hive query on a single table can be as fast as running the query in the Phoenix CLI with the following property settings: hive.fetch.task.conversion=more and hive.exec.parallel=true

Limitations

• Hive update and delete operations require transaction manager support on both Hive and Phoenix sides. Related Hive and Phoenix JIRAs are listed in the Resources section.
• Column mapping does not work correctly with mapping row key columns.
• MapReduce and Tez jobs always have a single reducer.

Resources

• PHOENIX-2743: Implementation accepted by the Apache Phoenix community. Original pull request contains modifications for Hive classes.
• PHOENIX-331: An outdated implementation with support for Hive 0.98.

Pig Integration

Pig integration may be divided into two parts: a StoreFunc as a means to generate Phoenix-encoded data through Pig, and a Loader which enables Phoenix-encoded data to be read by Pig.

Pig StoreFunc

The StoreFunc allows users to write data in Phoenix-encoded format to HBase tables using Pig scripts. This is a nice way to bulk upload data from a MapReduce job in parallel to a Phoenix table in HBase. All you need to specify is the endpoint address, HBase table name and a batch size. For example:

A = load 'testdata' as (a:chararray, b:chararray, c:chararray, d:chararray, e:datetime);
STORE A into 'hbase://CORE.ENTITY_HISTORY' using
    org.apache.phoenix.pig.PhoenixHBaseStorage('localhost', '-batchSize 5000');

The above reads a file testdata and writes the elements to table CORE.ENTITY_HISTORY in HBase running on localhost. The first StoreFunc argument is the server, and the second is the batch size for Phoenix upserts. The batch size is related to how many rows you can hold in memory. A good default is 1000 rows, but if your rows are wide, you may want to decrease this.

Note that Pig types must be in sync with the target Phoenix data types. This StoreFunc tries best to cast based on input Pig types and target Phoenix data types, but it is recommended to provide an appropriate schema.

Gotchas

It is advised that the upsert operation be idempotent. That is, trying to re-upsert data should not cause any inconsistencies. This is important in the case when a Pig job fails in process of writing to a Phoenix table. There is no notion of rollback (due to lack of transactions in HBase), and re-trying the upsert with PhoenixHBaseStorage must result in the same data in HBase table.

For example, let’s assume we are writing records n1...n10 to HBase. If the job fails in the middle of this process, we are left in an inconsistent state where n1...n7 made it to the phoenix tables but n8...n10 were missed. If we retry the same operation, n1...n7 would be re-upserted and n8...n10 would be upserted this time.

Pig Loader

A Pig data loader allows users to read data from Phoenix-backed HBase tables within a Pig script.

The LoadFunc provides two alternative ways to load data.

1. Given a table name, the following will load the data for all the columns in the HIRES table:

   A = load 'hbase://table/HIRES' using org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');

   To restrict the list of columns, you may specify the column names as part of LOAD as shown below:

   A = load 'hbase://table/HIRES/ID,NAME' using org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');

   Here, only data for ID and NAME columns are returned.

2. Given a query, the following loads data for all those rows whose AGE column has a value of greater than 50:

   A = load 'hbase://query/SELECT ID,NAME FROM HIRES WHERE AGE > 50' using org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');

   The LOAD func merely executes the given SQL query and returns the results. Though there is a provision to provide a query as part of LOAD, it is restricted to the following:

   • Only a SELECT query is allowed. No DML statements such as UPSERT or DELETE.
   • The query may not contain any GROUP BY, ORDER BY, LIMIT, or DISTINCT clauses.
   • The query may not contain any AGGREGATE functions.

In both cases, the ZooKeeper quorum should be passed to PhoenixHBaseLoader as a constructor argument.

The LoadFunc makes a best effort to map Phoenix data types to Pig datatypes. You can check org.apache.phoenix.pig.util.TypeUtil to see how each Phoenix data type maps to Pig.

Example

Determine the number of users by a CLIENT ID.

DDL

CREATE TABLE HIRES (
  CLIENTID INTEGER NOT NULL,
  EMPID INTEGER NOT NULL,
  NAME VARCHAR
  CONSTRAINT pk PRIMARY KEY(CLIENTID, EMPID)
);

Pig Script

raw = LOAD 'hbase://table/HIRES' USING org.apache.phoenix.pig.PhoenixHBaseLoader('localhost');
grpd = GROUP raw BY CLIENTID;
cnt = FOREACH grpd GENERATE group AS CLIENT, COUNT(raw);
DUMP cnt;

Future Work

• Support for ARRAY data type.
• Usage of expressions within the SELECT clause when providing a full query.

Map Reduce Integration

Phoenix provides support for retrieving and writing to Phoenix tables from within MapReduce jobs. The framework now provides custom InputFormat and OutputFormat classes PhoenixInputFormat , PhoenixOutputFormat.

PhoenixMapReduceUtil provides several utility methods to set the input and output configuration parameters to the job.

When a Phoenix table is the source for the MapReduce job, we can provide a SELECT query or pass a table name and specific columns to import data. To retrieve data from the table within the mapper class, we need to have a class that implements DBWritable and pass it as an argument to PhoenixMapReduceUtil.setInput. The custom DBWritable class provides an implementation for readFields(ResultSet rs) that allows us to retrieve columns for each row. This custom DBWritable class will form the input value to the mapper class.

Note: The SELECT query must not perform any aggregation or use DISTINCT as these are not supported by our map-reduce integration.

Similarly, when writing to a Phoenix table, we use PhoenixMapReduceUtil.setOutput to set the output table and columns.

Note: Phoenix internally builds the UPSERT query for you .

The output key and value class for the job should always be NullWritable and the custom DBWritable class that implements the write method.

Let's dive into an example where we have a table, STOCK, that holds the master data of quarterly recordings in a double array for each year, and we want to find the max price of each stock across all years. Let's store the output in STOCK_STATS, which is another Phoenix table.

Note: You can definitely configure a job to read from HDFS and load into a Phoenix table.

a) stock

CREATE TABLE IF NOT EXISTS STOCK (
  STOCK_NAME VARCHAR NOT NULL,
  RECORDING_YEAR INTEGER NOT NULL,
  RECORDINGS_QUARTER DOUBLE ARRAY[],
  CONSTRAINT pk PRIMARY KEY (STOCK_NAME, RECORDING_YEAR)
);

b) stock_stats

CREATE TABLE IF NOT EXISTS STOCK_STATS (
  STOCK_NAME VARCHAR NOT NULL,
  MAX_RECORDING DOUBLE,
  CONSTRAINT pk PRIMARY KEY (STOCK_NAME)
);

Sample Data

UPSERT into STOCK values ('AAPL',2009,ARRAY[85.88,91.04,88.5,90.3]);
UPSERT into STOCK values ('AAPL',2008,ARRAY[199.27,200.26,192.55,194.84]);
UPSERT into STOCK values ('AAPL',2007,ARRAY[86.29,86.58,81.90,83.80]);
UPSERT into STOCK values ('CSCO',2009,ARRAY[16.41,17.00,16.25,16.96]);
UPSERT into STOCK values ('CSCO',2008,ARRAY[27.00,27.30,26.21,26.54]);
UPSERT into STOCK values ('CSCO',2007,ARRAY[27.46,27.98,27.33,27.73]);
UPSERT into STOCK values ('CSCO',2006,ARRAY[17.21,17.49,17.18,17.45]);
UPSERT into STOCK values ('GOOG',2009,ARRAY[308.60,321.82,305.50,321.32]);
UPSERT into STOCK values ('GOOG',2008,ARRAY[692.87,697.37,677.73,685.19]);
UPSERT into STOCK values ('GOOG',2007,ARRAY[466.00,476.66,461.11,467.59]);
UPSERT into STOCK values ('GOOG',2006,ARRAY[422.52,435.67,418.22,435.23]);
UPSERT into STOCK values ('MSFT',2009,ARRAY[19.53,20.40,19.37,20.33]);
UPSERT into STOCK values ('MSFT',2008,ARRAY[35.79,35.96,35.00,35.22]);
UPSERT into STOCK values ('MSFT',2007,ARRAY[29.91,30.25,29.40,29.86]);
UPSERT into STOCK values ('MSFT',2006,ARRAY[26.25,27.00,26.10,26.84]);
UPSERT into STOCK values ('YHOO',2009,ARRAY[12.17,12.85,12.12,12.85]);
UPSERT into STOCK values ('YHOO',2008,ARRAY[23.80,24.15,23.60,23.72]);
UPSERT into STOCK values ('YHOO',2007,ARRAY[25.85,26.26,25.26,25.61]);
UPSERT into STOCK values ('YHOO',2006,ARRAY[39.69,41.22,38.79,40.91]);

Below is a simple job configuration

Job Configuration

final Configuration configuration = HBaseConfiguration.create();
final Job job = Job.getInstance(configuration, "phoenix-mr-job");

// We can either specify a selectQuery or ignore it when we would like to retrieve all the columns
final String selectQuery = "SELECT STOCK_NAME,RECORDING_YEAR,RECORDINGS_QUARTER FROM STOCK ";

// StockWritable is the DBWritable class that enables us to process the result of the above query
PhoenixMapReduceUtil.setInput(job, StockWritable.class, "STOCK", selectQuery);

// Set the target Phoenix table and the columns
PhoenixMapReduceUtil.setOutput(job, "STOCK_STATS", "STOCK_NAME,MAX_RECORDING");

job.setMapperClass(StockMapper.class);
job.setReducerClass(StockReducer.class);
job.setOutputFormatClass(PhoenixOutputFormat.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(DoubleWritable.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(StockWritable.class);
TableMapReduceUtil.addDependencyJars(job);
job.waitForCompletion(true);

StockWritable

public class StockWritable implements DBWritable, Writable {
  private String stockName;
  private int year;
  private double[] recordings;
  private double maxPrice;

  @Override
  public void readFields(DataInput input) throws IOException {
  }

  @Override
  public void write(DataOutput output) throws IOException {
  }

  @Override
  public void readFields(ResultSet rs) throws SQLException {
    stockName = rs.getString("STOCK_NAME");
    year = rs.getInt("RECORDING_YEAR");
    final Array recordingsArray = rs.getArray("RECORDINGS_QUARTER");
    recordings = (double[]) recordingsArray.getArray();
  }

  @Override
  public void write(PreparedStatement pstmt) throws SQLException {
    pstmt.setString(1, stockName);
    pstmt.setDouble(2, maxPrice);
  }

  // getters / setters for the fields
  ...
  ...
}

Stock Mapper

public static class StockMapper extends Mapper<NullWritable, StockWritable, Text, DoubleWritable> {
  private Text stock = new Text();
  private DoubleWritable price = new DoubleWritable();

  @Override
  protected void map(NullWritable key, StockWritable stockWritable, Context context)
      throws IOException, InterruptedException {
    double[] recordings = stockWritable.getRecordings();
    final String stockName = stockWritable.getStockName();
    double maxPrice = Double.MIN_VALUE;
    for (double recording : recordings) {
      if (maxPrice < recording) {
        maxPrice = recording;
      }
    }
    stock.set(stockName);
    price.set(maxPrice);
    context.write(stock, price);
  }
}

Stock Reducer

public static class StockReducer extends Reducer<Text, DoubleWritable, NullWritable, StockWritable> {
  @Override
  protected void reduce(Text key, Iterable<DoubleWritable> recordings, Context context)
      throws IOException, InterruptedException {
    double maxPrice = Double.MIN_VALUE;
    for (DoubleWritable recording : recordings) {
      if (maxPrice < recording.get()) {
        maxPrice = recording.get();
      }
    }
    final StockWritable stock = new StockWritable();
    stock.setStockName(key.toString());
    stock.setMaxPrice(maxPrice);
    context.write(NullWritable.get(), stock);
  }
}

Packaging and Running

1. Ensure phoenix-[version]-client.jar is in the classpath of your MapReduce job JAR.
2. To run the job, use the hadoop jar command with the necessary arguments.

Flume Plugin

The plugin enables reliable and efficient streaming of large amounts of data/logs into HBase using the Phoenix API. The custom Phoenix sink and event serializer must be configured in the Flume agent configuration file. Currently, the only supported event serializer is RegexEventSerializer, which parses the Flume event body using a configured regex.

Prerequisites

• Phoenix v3.0.0-SNAPSHOT+
• Flume 1.4.0+

Installation and Setup

1. Download and build Phoenix v3.0.0-SNAPSHOT.
2. Follow the instructions here to build the project, as the Flume plugin is still under beta.
3. Create a plugins.d directory within $FLUME_HOME. Within that, create sub-directory phoenix-sink/lib.
4. Copy generated phoenix-3.0.0-SNAPSHOT-client.jar to $FLUME_HOME/plugins.d/phoenix-sink/lib.

Configuration

For an example configuration for ingesting Apache access logs into Phoenix, see this property file. It uses UUID as a row key generator for the primary key.

Starting the agent

$ bin/flume-ng agent -f conf/flume-conf.properties -c ./conf -n agent

Monitoring

To monitor the agent and sink process, enable JMX via flume-env.sh ($FLUME_HOME/conf/flume-env.sh). Ensure you have the following line uncommented:

JAVA_OPTS="-Xms1g -Xmx1g -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=3141 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false"

Kafka Plugin

The plugin enables reliable and efficient streaming of large amounts of data/logs into HBase using the Phoenix API.

Apache Kafka™ is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.

At a high level, producers send messages over the network to the Kafka cluster, which then serves them to consumers:

Phoenix provides PhoenixConsumer to receive messages from Kafka producers.

Prerequisites

• Phoenix 4.10.0+
• Kafka 0.9.0.0+

Installation and Setup

Use our binary artifacts for Phoenix 4.10.0+ directly or download and build Phoenix yourself (see instructions here).

PhoenixConsumer with RegexEventSerializer

Create a kafka-consumer-regex.properties file with the following properties:

serializer=regex
serializer.rowkeyType=uuid
serializer.regex=([^\,]*),([^\,]*),([^\,]*)
serializer.columns=c1,c2,c3

jdbcUrl=jdbc:phoenix:localhost
table=SAMPLE1
ddl=CREATE TABLE IF NOT EXISTS SAMPLE1(uid VARCHAR NOT NULL,c1 VARCHAR,c2 VARCHAR,c3 VARCHAR CONSTRAINT pk PRIMARY KEY(uid))

bootstrap.servers=localhost:9092
topics=topic1,topic2
poll.timeout.ms=100

PhoenixConsumer with JsonEventSerializer

Create a kafka-consumer-json.properties file with the following properties:

serializer=json
serializer.rowkeyType=uuid
serializer.columns=c1,c2,c3

jdbcUrl=jdbc:phoenix:localhost
table=SAMPLE2
ddl=CREATE TABLE IF NOT EXISTS SAMPLE2(uid VARCHAR NOT NULL,c1 VARCHAR,c2 VARCHAR,c3 VARCHAR CONSTRAINT pk PRIMARY KEY(uid))

bootstrap.servers=localhost:9092
topics=topic1,topic2
poll.timeout.ms=100

PhoenixConsumer Execution Procedure

Start the Kafka producer, then send some messages:

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1

Learn more about Apache Kafka here.

Start PhoenixConsumer using the command below:

HADOOP_CLASSPATH=$(hbase classpath):/path/to/hbase/conf hadoop jar phoenix-kafka-<version>-minimal.jar org.apache.phoenix.kafka.consumer.PhoenixConsumerTool --file /data/kafka-consumer.properties

The input file must be present on HDFS (not the local filesystem where the command is being run).

Configuration

Note: This plugin supports Phoenix Flume event serializers.

• RegexEventSerializer parses Kafka messages based on the regex specified in the configuration file.
• JsonEventSerializer parses Kafka messages based on the schema specified in the configuration file.

Python Driver

The Python driver for Apache Phoenix implements the Python DB 2.0 API to access Phoenix via Phoenix Query Server. The driver is tested with Python 2.7 and 3.5-3.8. This code was originally called python-phoenixdb and was graciously donated by its authors to the Apache Phoenix project.

All future development of the project is being done in Apache Phoenix.

Installation

From PyPI

The latest release is always available from PyPI, and can be installed by pip/pip3 as usual.

pip3 install --user phoenixdb

From source

You can build phoenixdb from the official source release, or use the latest development version from the source repository. The python-phoenixdb source lives in the python-phoenixdb directory of the phoenix-queryserver repository.

$ cd python-phoenixdb # (Only when building from the git repo)
$ pip install -r requirements.txt
$ python setup.py install

Examples

import phoenixdb
import phoenixdb.cursor

database_url = 'http://localhost:8765/'
conn = phoenixdb.connect(database_url, autocommit=True)

cursor = conn.cursor()
cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, username VARCHAR)")
cursor.execute("UPSERT INTO users VALUES (?, ?)", (1, 'admin'))
cursor.execute("SELECT * FROM users")
print(cursor.fetchall())

cursor = conn.cursor(cursor_factory=phoenixdb.cursor.DictCursor)
cursor.execute("SELECT * FROM users WHERE id=1")
print(cursor.fetchone()['USERNAME'])

Limitations

• None presently known.

Resources

• PHOENIX-4636: Initial landing of the driver into Apache Phoenix.
• PHOENIX-4688: Implementation of Kerberos authentication via SPNEGO.

Addons

Phoenix ORM library

PHO is an Object Relational Mapping (ORM) library for building and executing queries on HBase using Apache Phoenix. It provides ORM-style mappings and DSL-style query building. Initially developed and open sourced by eHarmony, it is available on GitHub.

Its interfaces and generic annotations make it possible to switch data store APIs in the future without changing query definitions. Currently, it supports HBase integration through Apache Phoenix, and can be extended with other implementations.

Entity Class

Suppose we have the following TestClass we want to query in our data store:

// class must be annotated with Entity
import com.google.code.morphia.annotations.Embedded;
import com.google.code.morphia.annotations.Entity;

@Entity(value = "user_matches")
public class MatchDataFeedItemDto {
    @Embedded
    private MatchCommunicationElement communication;
    @Embedded
    private MatchElement match;
    @Embedded
    private MatchProfileElement matchedUser;
}

public class MatchElement {
    // row key
    @Property(value = "UID")
    private long userId;
    @Property(value = "MID")
    private long matchId;
    @Property(value = "DLVRYDT")
    private Date deliveredDate;
    @Property(value = "STATUS")
    private int status;
}

Query Building

Query building can be done in DSL style. More advanced query building is under development, but for now we will use a combination of QueryBuilder and static, Hibernate-style Restrictions methods to construct queries.

Simple queries

Construct a query to find all user matches delivered in the past two days and not in a closed state.

import com.eharmony.datastore.api.DataStoreApi;
import com.eharmony.datastore.model.MatchDataFeedItemDto;
import com.eharmony.datastore.query.QuerySelect;
import com.eharmony.datastore.query.builder.QueryBuilder;
import com.eharmony.datastore.query.criterion.Restrictions;

@Repository
public class MatchStoreQueryRepositoryImpl implements MatchStoreQueryRepository {
    final QuerySelect<MatchDataFeedItemDto, MatchDataFeedItemDto> query = QueryBuilder
        .builderFor(MatchDataFeedItemDto.class)
        .select()
        .add(Restrictions.eq("userId", userId))
        .add(Restrictions.eq("status", 2))
        .add(Restrictions.gt("deliveredDate", timeThreshold.getTime()))
        .build();

    Iterable<MatchDataFeedItemDto> feedItems = dataStoreApi.findAll(query);
}

Compound queries

Construct a more complex query where we find items older than one day, include multiple status values, order by deliveryDate, and limit result size to 10.

// provided
List<Integer> statusFilters = request.getMatchStatusFilters();
String sortBy = request.getSortBy();
Disjunction disjunction = new Disjunction();

for (Integer statusFilter : statusFilters) {
    disjunction.add(Restrictions.eq("status", statusFilter));
}

final QuerySelect<MatchDataFeedItemDto, MatchDataFeedItemDto> query = QueryBuilder
    .builderFor(MatchDataFeedItemDto.class)
    .select()
    .add(Restrictions.eq("userId", userId))
    .add(Restrictions.gt("deliveredDate", timeThreshold.getTime()))
    .add(disjunction)
    .addOrder(new Ordering(sortBy, Order.DESCENDING))
    .build();

Iterable<MatchDataFeedItemDto> feedItems = dataStoreApi.findAll(query);

By default, expressions are combined with AND when added separately.

Query Interface

The following query components are supported:

// equals
EqualityExpression eq(String propertyName, Object value);

// does not equal
EqualityExpression ne(String propertyName, Object value);

// less than
EqualityExpression lt(String propertyName, Object value);

// less than or equal
EqualityExpression lte(String propertyName, Object value);

// greater than
EqualityExpression gt(String propertyName, Object value);

// greater than or equal
EqualityExpression gte(String propertyName, Object value);

// between from and to (inclusive)
RangeExpression between(String propertyName, Object from, Object to);

// and - takes a variable list of expressions as arguments
Conjunction and(Criterion... criteria);

// or - takes a variable list of expressions as arguments
Disjunction or(Criterion... criteria);