pytd 1.5.2

Query in Treasure Data

Issue Presto query and retrieve the result:

client.query('select symbol, count(1) as cnt from nasdaq group by 1 order by 1')
# {'columns': ['symbol', 'cnt'], 'data': [['AAIT', 590], ['AAL', 82], ['AAME', 9252], ..., ['ZUMZ', 2364]]}

In case of Hive:

client.query('select hivemall_version()', engine='hive')
# {'columns': ['_c0'], 'data': [['0.6.0-SNAPSHOT-201901-r01']]} (as of Feb, 2019)

It is also possible to explicitly initialize pytd.Client for Hive:

client_hive = pytd.Client(database='sample_datasets', default_engine='hive')
client_hive.query('select hivemall_version()')

Here is an example of generator-based iterative retrieval using DB-API. For details, please refer to Documentation

from pytd.dbapi import connect

conn = connect(pytd.Client(database='sample_datasets'))
# or, connect with Hive:
# >>> conn = connect(pytd.Client(database='sample_datasets', default_engine='hive'))

def iterrows(sql, connection):
   cur = connection.cursor()
   cur.execute(sql)
   index = 0
   columns = None
   while True:
      row = cur.fetchone()
      if row is None:
         break
      if columns is None:
         columns = [desc[0] for desc in cur.description]
      yield index, dict(zip(columns, row))
      index += 1

for index, row in iterrows('select symbol, count(1) as cnt from nasdaq group by 1 order by 1', conn):
   print(index, row)

When you face unexpected timeout error with Presto, you can try iterative way to retrieve data.

Write data to Treasure Data

Data represented as pandas.DataFrame can be written to Treasure Data as follows:

import pandas as pd

df = pd.DataFrame(data={'col1': [1, 2], 'col2': [3, 10]})
client.load_table_from_dataframe(df, 'takuti.foo', writer='bulk_import', if_exists='overwrite')

For the writer option, pytd supports three different ways to ingest data to Treasure Data:

1. Bulk Import API: bulk_import (default)

   • Convert data into a CSV file and upload in the batch fashion.

2. Presto INSERT INTO query: insert_into

   • Insert every single row in DataFrame by issuing an INSERT INTO query through the Presto query engine.

   • Recommended only for a small volume of data.

3. td-spark: spark (No longer available)

   • Local customized Spark instance directly writes DataFrame to Treasure Data’s primary storage system.

Characteristics of each of these methods can be summarized as follows:

from pytd.writer import SparkWriter

writer = SparkWriter(td_spark_path='/path/to/td-spark-assembly.jar')
client.load_table_from_dataframe(df, 'mydb.bar', writer=writer, if_exists='overwrite')

Comparison between pytd, td-client-python, and pandas-td

Treasure Data offers three different Python clients on GitHub, and the following list summarizes their characteristics.

1. td-client-python

   • Basic REST API wrapper.

   • Similar functionalities to td-client-{ruby, java, node, go}.

   • The capability is limited by what Treasure Data REST API can do.

2. pytd

   • Access to Plazma via td-spark as introduced above.

   • Efficient connection to Presto based on presto-python-client.

   • Multiple data ingestion methods and a variety of utility functions.

3. pandas-td (deprecated)

   • Old tool optimized for pandas and Jupyter Notebook.

   • pytd offers its compatible function set (see below for the detail).

An optimal choice of package depends on your specific use case, but common guidelines can be listed as follows:

• Use td-client-python if you want to execute basic CRUD operations from Python applications.

• Use pytd for (1) analytical purpose relying on pandas and Jupyter Notebook, and (2) achieving more efficient data access at ease.

• Do not use pandas-td. If you are using pandas-td, replace the code with pytd based on the following guidance as soon as possible.

pip install pytd
# or, `pip install pytd[spark]` if you wish to use `to_td`

import pandas_td as td

In [1]: %%load_ext pandas_td.ipython

import pytd.pandas_td as td

In [1]: %%load_ext pytd.pandas_td.ipython