# Performing SQL, Hive and Impala queries[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#performing-sql-hive-and-impala-queries "Permalink to this headline")

You can use the Python APIs to execute SQL queries on any SQL connection in DSS (including Hive and Impala).

* Executing queries

+ Queries with side-effects

* Partial recipes

* Executing queries (dataikuapi variant)

* Reference documentation

Note

There are three capabilities related to performing SQL queries in Dataiku’s Python APIs:

* `dataiku.SQLExecutor2`, `dataiku.HiveExecutor` and `dataiku.ImpalaExecutor` in the `dataiku` package. It was initially designed for usage within DSS in recipes and Jupyter notebooks. These are used to perform queries and retrieve results, either as an iterator or as a pandas dataframe

* “partial recipes”. It is possible to execute a “partial recipe” from a Python recipe, to execute a Hive, Impala or SQL query. This allows you to use Python to dynamically generate a SQL (resp Hive, Pig, Impala) query and have DSS execute it, as if your recipe was a SQL query recipe. This is useful when you need complex business logic to generate the final SQL query and can’t do it with only SQL constructs.

* `dataikuapi.DSSClient.sql\_query()` in the `dataikuapi` package. This function was initially designed for usage outside of DSS and only supports returning results as an iterator. It does not support pandas dataframe

We recommend the usage of the `dataiku` variants.

For more details on the two packages, please see Python APIs

## Executing queries[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#executing-queries "Permalink to this headline")

You can retrieve the results of a SELECT query as a Pandas dataframe.

§ from dataiku import SQLExecutor2

§ executor = SQLExecutor2(connection="db-connection") # or dataset="dataset\_name"

§ df = executor.query\_to\_df("SELECT col1, COUNT(\*) as count FROM mytable")

§ # df is a Pandas dataframe with two columns : "col1" and "count"

Alternatively, you can retrieve the results of a query as an iterator.

§ from dataiku import SQLExecutor2

§ executor = SQLExecutor2(connection="db-connection")

§ query\_reader = executor.query\_to\_iter("SELECT \* FROM mytable")

§ query\_iterator = query\_reader.iter\_tuples()

### Queries with side-effects[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#queries-with-side-effects "Permalink to this headline")

For databases supporting commit, the transaction in which the queries are executed is rolled back at the end, as is the default in DSS.

In order to perform queries with side-effects such as INSERT or UPDATE, you need to add `post\_queries=['COMMIT']` to your `query\_to\_df` call.

Depending on your database, DDL queries such as `CREATE TABLE` will also need a `COMMIT` or not.

## Partial recipes[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#partial-recipes "Permalink to this headline")

It is possible to execute a “partial recipe” from a Python recipe, to execute a Hive, Pig, Impala or SQL query.

This allows you to use Python to dynamically generate a SQL (resp Hive, Pig, Impala) query and have DSS execute it, as if your recipe was a SQL query recipe.

This is useful when you need complex business logic to generate the final SQL query and can’t do it with only SQL constructs.

Note

Partial recipes are only possible when you are running a Python recipe. It is not available in the notebooks nor outside of DSS.

The partial recipe behaves like the corresponding SQL (resp Hive, Impala) recipe w.r.t. the inputs and outputs. Notably, a Python recipe in which a partial Hive recipe is executed can only have HDFS datasets as inputs and outputs. Likewise, a Impala or SQL partial recipe having only one ouput, the output dataset has to be specified for the partial recipe execution.

In the following example, we make a first query in order to dynamically build the larger query that runs as the “main” query of the recipe.

§ from dataiku import SQLExecutor2

§ # get the needed data to prepare the query

§ # for example, load from another table

§ executor = SQLExecutor2(dataset=my\_auxiliary\_dataset)

§ words = executor.query\_to\_df(

§ "SELECT word FROM word\_frequency WHERE frequency > 0.01 AND frequency < 0.99")

§ # prepare a query dynamically

§ sql = 'SELECT id '

§ for word in words['word']:

§ sql = sql + ", (length(text) - length(regexp\_replace(text, '" + word + "', ''))) / " + len(word) + " AS count\_" + word

§ sql = sql + " FROM reviews"

§ # execute it

§ # no need to instantiate an executor object, the method is static

§ my\_output\_dataset = dataiku.Dataset("my\_output\_dataset\_name")

§ SQLExecutor2.exec\_recipe\_fragment(my\_output\_dataset, sql)

## Executing queries (dataikuapi variant)[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#executing-queries-dataikuapi-variant "Permalink to this headline")

Note

We recommend using `SQLExecutor2` rather, especially inside DSS.

Running a query against DSS is a 3-step process:

* create the query

* run it and fetch the data

* verify that the streaming of the results wasn’t interrupted

The verification will make DSS release the resources taken for the query’s execution, so the verify() call has to be done once the results have been streamed.

An example of a SQL query on a connection configured in DSS is:

§ streamed\_query = client.sql\_query('select \* from train\_set', connection='local\_postgres', type='sql')

§ row\_count = 0

§ for row in streamed\_query.iter\_rows():

§ row\_count = row\_count + 1

§ streamed\_query.verify() # raises an exception in case something went wrong

§ print('the query returned %i rows' % count)

Queries against Hive and Impala are also possible. In that case, the type must be set to ‘hive’ or ‘impala’ accordingly, and instead of a connection it is possible to pass a database name:

§ client = DSSClient(host, apiKey)

§ streamed\_query = client.sql\_query('select \* from train\_set', database='test\_area', type='impala')

§ ...

In order to run queries before or after the main query, but still in the same session, for example to set variables in the session, the API provides 2 parameters pre\_queries and post\_queries which take in arrays of queries:

§ streamed\_query = client.sql\_query('select \* from train\_set', database='test\_area', type='hive', pre\_queries=['set hive.execution.engine=tez'])

§ ...

## Reference documentation[¶](https://doc.dataiku.com/dss/latest/api/python/sql.html#reference-documentation "Permalink to this headline")

*class* `dataiku.``SQLExecutor2`(*connection=None*, *dataset=None*)

This is a handle to execute SQL statements on a given SQL connection.

The connection is derived from either the connection parameter or the dataset parameter.

* Parameters: * **connection** (*string*) – name of the SQL connection
* **dataset** – name of a dataset or a
`dataiku.Dataset` object.

*static* `exec_recipe_fragment`(*output\_dataset*, *query*, *pre\_queries=[]*, *post\_queries=[]*, *overwrite\_output\_schema=True*, *drop\_partitioned\_on\_schema\_mismatch=False*)

Executes a SQL query and store the results to the output\_dataset after dropping its underlying table.

* Parameters: * **output\_dataset** (*object*) – `dataiku.Dataset` output dataset where
to write the result of the query.
* **query** (*str*) – SQL main query
* **pre\_queries** (*list*) – list of queries to be executed before
the main query
* **post\_queries** (*list*) – list of queries to be executed after
the main query
* **overwrite\_output\_schema** (*bool*) – if True, generates the output
schema from the query results. If False, maintains the existing
output schema
* **drop\_partitioned\_on\_schema\_mismatch** (*bool*) – for partitioned
output datasets. If True, drops all partitions whose schema is
inconsistent with that of the dataset. Only relevant
when overwrite\_output\_schema=True

* Returns: None

`query_to_df`(*query*, *pre\_queries=None*, *post\_queries=None*, *extra\_conf={}*, *infer\_from\_schema=False*, *parse\_dates=True*, *bool\_as\_str=False*, *dtypes=None*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)

This function returns the result of the main query as a pandas dataframe.

* Parameters: * **query** (*str*) – SQL main query
* **pre\_queries** (*list*) – list of queries to be executed before the main query
* **post\_queries** (*list*) – list of queries to be executed after the main query
* **extra\_conf** – do not use
* **infer\_from\_schema** (*bool*) – if True, the resulting pandas dataframe types
are set per the SQL query datatypes rather than being inferred by pandas
* **parse\_dates** (*bool*) – if True, SQL datetime columns are set as datetime
dtypes in the resulting pandas dataframe. The infer\_from\_schema must be True
for this param to be relevant
* **bool\_as\_str** (*bool*) – whether to cast boolean values as string
* **dtypes** (*dict*) – with key= column name and value=`numpy.dtype()`
* **script\_steps** – do not use
* **script\_input\_schema** – do not use
* **script\_output\_schema** – do not use

* Returns: a pandas dataframe with the result of the query.

`query_to_iter`(*query*, *pre\_queries=None*, *post\_queries=None*, *extra\_conf={}*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)

This function returns a `QueryReader` to iterate on the rows.

* Parameters: * **query** (*str*) – the main query
* **pre\_queries** (*list*) – list of queries to be executed before the main query
* **post\_queries** (*list*) – list of queries to be executed after the main query
* **script\_steps** – do not use
* **script\_input\_schema** – do not use
* **script\_output\_schema** – do not use

* Returns: a `QueryReader` to iterate on the rows.

*class* `dataiku.``HiveExecutor`(*dataset=None*, *database=None*, *connection=None*)

*static* `exec_recipe_fragment`(*query*, *pre\_queries=[]*, *post\_queries=[]*, *overwrite\_output\_schema=True*, *drop\_partitioned\_on\_schema\_mismatch=False*, *metastore\_handling=None*, *extra\_conf={}*, *add\_dku\_udf=False*)

`query_to_df`(*query*, *pre\_queries=None*, *post\_queries=None*, *extra\_conf={}*, *infer\_from\_schema=False*, *parse\_dates=True*, *bool\_as\_str=False*, *dtypes=None*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)

`query_to_iter`(*query*, *pre\_queries=None*, *post\_queries=None*, *extra\_conf={}*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)

*class* `dataiku.``ImpalaExecutor`(*dataset=None*, *database=None*, *connection=None*)

*static* `exec_recipe_fragment`(*output\_dataset*, *query*, *pre\_queries=[]*, *post\_queries=[]*, *overwrite\_output\_schema=True*, *use\_stream\_mode=True*)

`query_to_df`(*query*, *pre\_queries=None*, *post\_queries=None*, *connection=None*, *extra\_conf={}*, *infer\_from\_schema=False*, *parse\_dates=True*, *bool\_as\_str=False*, *dtypes=None*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)

`query_to_iter`(*query*, *pre\_queries=None*, *post\_queries=None*, *connection=None*, *extra\_conf={}*, *script\_steps=None*, *script\_input\_schema=None*, *script\_output\_schema=None*)
