<commit_before>---
layout: post
title: "Integrating Hive UDF's in Python"
excerpt_separator: <!--excerpt-->
tags: [python, hive]
---

I need to do some pretty flexible things in my Hive queries, so flexible
that it's beyond the capability of Hive QL.
Writing a Hive UDF (user defined function) is an option.
However, all the online examples I can find require the UDF to be a standing-alone script,
placed at a known location in HDFS, and used via the `ADD FILE` clause that is understood by the Hive CLI.
Having to put the script in HDFS and use it on a machine that has the Hive CLI installed means interruption to my Python code flow, which I hate.
I want the Hive UDF to be seamlessly integrated into my Python code.
How can I achieve that?

## Preparations

Before describing my solution to UDF, I need to make two preparations:
the first is a Python client that easily interacts with Hive server, as long as UDF is not involved;
the second is a toy Hive table for testing.

The Python client is developed based on the package `impyla`.
Skipping all the details, the client is used like this:

```python
hive = Hive()

sql = '''
CREATE TABLE mytable (
    a INT, 
    b STRING, 
    c STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t'
LINES TERMINATED BY '\n'
STORED AS TEXTFILE'''

hive.write(sql)

sql = 'SELECT a, b, c FROM mydb.mytable'
hive.read(sql)
rows = hive.fetchall()
```

This can execute any valid query, and fetching results following a 'read' query,
without ever leaving the Python code flow.

The toy table, `mydb.cars` contains three columns: an integer `id`, and a JSON string `info_json`.
The following rows are inserted into the table:

```
1, '{"make": "honda", "price": 1000}'
2, '{"make": "ford", "price": 2000}'
3, '{"make": "ford"}'
4, '{"make": "tesla", "price": 3000}'
5, '{"make": "honda", "price": 2000}'
6, '{"make": "ford", "price": 4000}'
```

## The UDF syntax

The basic syntax for using a Hive UDF defined in script `myudf.py` is as follows:

```
ADD FILE hdfs:///tmp/myudf.py;

SELECT 
    TRANSFORM(id, info_json)
    USING 'myudf.py'
    AS (out STRING)
FROM (
    SELECT * FROM mydb.cars
) AS tmp
```

Here, the rows retrieved by `FROM (SELECT ...)` are processed by `myudf.py`, which prints out a single column, being called `out`. The input fields to `myudf.py` are given names by `TRANSFORM(...)`.

Let's make `myudf.py` very simple---it simply returns the second field:

```python
#!/usr/bin/env python
from __future__ import print_function
import sys

SEP = '\t'
NULL = '\\N'

def main():
    for line in sys.stdin:
        _, info_json = line.strip().split(SEP)
        print(info_json)


if __name__ == '__main__':
    main()
```

Here's how a UDF script works:

1. Input rows are fed to the script on `stdin`.
2. The script takes each input row and prints out something, which is taken as the output.
   If the print-out contains the field separator, then it is interpreted as multiple columns.
   In the example above, the print-out contains a single column, which is named `out`
   in the HiveQL statement that uses this UDF.
3. Because print-outs to `stdout` are taken as the result, there is no way to enforce that
   each input line has to lead to an output line.
   If the script prints multiple lines for any, or each, input line, they are simply all result lines.
   If the script does not print anything while processing a certain input line,
   that amounts to filtering out that input row in the query.

I came to the following critical realization (which is not emphasized in any of the online tutorials I found):

> The `xyz` specified by `USING 'xyz'` must be a program that can run on the Hive server.
> It should take input and print out output as described above.
> The condition 'can run' also implies it can be found, e.g. it's on the system `PATH`.
>
> And this is all that needs to be satisfied.
> It does not matter what language the script is written in---e.g. it's perfectly fine if it's a binary executable produced by some `C++` code---as long as it just runs.
> The `ADD FILE` statement is not essential---again, as long as it runs.

Let's verify it. Let's make a UDF that is simply `echo xyz`.
This UDF ignores any input, and just prints out a single line of output, `xyz`.
I logged into a machine that has the `hive` CLI, and did this:

```
$ hive
hive> use mydb;
OK
Time taken: 0.014 seconds
hive> SELECT TRANSFORM(id, info_json) USING 'echo xyz' AS (out STRING) FROM (SELECT * FROM cars) AS tmp;
Total jobs = 1
Launching Job 1 out of 1
... ... # ommitted
Total MapReduce CPU Time Spent: 1 seconds 940 msec
OK
xyz
Time taken: 47.852 seconds, Fetched: 1 row(s)
hive>
```

As expected, the 'program' `echo xyz` was run and printed a single row

```
xyz
```

This points in the direction of effort: if I can somehow include the entire UDF in the string '...' after `USING`, then I don't need to deal with creating a script, putting it in HDFS, and so on. It should just work!

Let the journey begin.<commit_msg>minor<commit_after>---
layout: post
title: "Integrating Hive UDF's in Python"
excerpt_separator: <!--excerpt-->
tags: [python, hive]
---

I need to do some pretty flexible things in my Hive queries, so flexible
that it's beyond the capability of Hive QL.
Writing a Hive UDF (user defined function) is an option.
However, all the online examples I can find require the UDF to be a standing-alone script,
placed at a known location in HDFS, and used via the `ADD FILE` clause that is understood by the Hive CLI.
Having to put the script in HDFS and use it on a machine that has the Hive CLI installed means interruption to my Python code flow, which I hate.
I want the Hive UDF to be seamlessly integrated into my Python code.
How can I achieve that?

## Preparations

Before describing my solution to UDF, I need to make two preparations:
the first is a Python client that easily interacts with Hive server, as long as UDF is not involved;
the second is a toy Hive table for testing.

The Python client is developed based on the package `impyla`.
Skipping all the details, the client is used like this:

```python
hive = Hive()

sql = '''
CREATE TABLE mytable (
    a INT, 
    b STRING, 
    c STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\t'
LINES TERMINATED BY '\n'
STORED AS TEXTFILE'''

hive.write(sql)

sql = 'SELECT a, b, c FROM mydb.mytable'
hive.read(sql)
rows = hive.fetchall()
```

This can execute any valid query, and fetching results following a 'read' query,
without ever leaving the Python code flow.

The toy table, `mydb.cars` contains three columns: an integer `id`, and a JSON string `info_json`.
The following rows are inserted into the table:

```
1, '{"make": "honda", "price": 1000}'
2, '{"make": "ford", "price": 2000}'
3, '{"make": "ford"}'
4, '{"make": "tesla", "price": 3000}'
5, '{"make": "honda", "price": 2000}'
6, '{"make": "ford", "price": 4000}'
```

## The UDF syntax

The basic syntax for using a Hive UDF defined in script `myudf.py` is as follows:

```
ADD FILE hdfs:///tmp/myudf.py;

SELECT 
    TRANSFORM(id, info_json)
    USING 'myudf.py'
    AS (out STRING)
FROM (
    SELECT * FROM mydb.cars
) AS tmp
```

Here, the rows retrieved by `FROM (SELECT ...)` are processed by `myudf.py`, which prints out a single column, being called `out`. The input fields to `myudf.py` are given names by `TRANSFORM(...)`.

Let's make `myudf.py` very simple---it simply returns the second field:

```python
#!/usr/bin/env python
from __future__ import print_function
import sys

SEP = '\t'
NULL = '\\N'

def main():
    for line in sys.stdin:
        _, info_json = line.strip().split(SEP)
        print(info_json)


if __name__ == '__main__':
    main()
```

Here's how a UDF script works:

1. Input rows are fed to the script on `stdin`.
2. The script takes each input row and prints out something, which is taken as the output.
   If the print-out contains the field separator, then it is interpreted as multiple columns.
   In the example above, the print-out contains a single column, which is named `out`
   in the HiveQL statement that uses this UDF.
3. Because print-outs to `stdout` are taken as the result, there is no way to enforce that
   each input line has to lead to an output line.
   If the script prints multiple lines for any, or each, input line, they are simply all result lines.
   If the script does not print anything while processing a certain input line,
   that amounts to filtering out that input row in the query.

I came to the following critical realization (which is not emphasized in any of the online tutorials I found):

> The `xyz` specified by `USING 'xyz'` must be a program that can run on the Hive server.
> It should take input and print out output as described above.
> The condition 'can run' also implies it can be found, e.g. it's on the system `PATH`.
>
> And this is all that needs to be satisfied.
> It does not matter what language the script is written in---e.g. it's perfectly fine if it's a binary executable produced by some `C++` code---as long as it just runs.
> The `ADD FILE` statement is not essential---again, as long as it runs.

Let's verify it. Let's make a UDF that is simply `echo xyz`.
This UDF ignores any input, and just prints out a single line of output, `xyz`.
I logged into a machine that has the `hive` CLI, and did this:

```
$ hive
hive> use mydb;
OK
Time taken: 0.014 seconds
hive> SELECT TRANSFORM(id, info_json) USING 'echo xyz' AS (out STRING) FROM (SELECT * FROM cars) AS tmp;
Total jobs = 1
Launching Job 1 out of 1
... ...
Total MapReduce CPU Time Spent: 1 seconds 940 msec
OK
xyz
Time taken: 47.852 seconds, Fetched: 1 row(s)
hive>
```

As expected, the 'program' `echo xyz` was run and printed a single row

```
xyz
```

This points in the direction of effort: if I can somehow include the entire UDF in the string '...' after `USING`, then I don't need to deal with creating a script, putting it in HDFS, and so on. It should just work!

Let the journey begin.

## Strategy of Python UDF

My goal is to write "in-line" Hive UDF in Python and send this code over to Hive server as part of the HiveQL statement, hence avoiding any script files. The way to run Python code without a script file is

```
$ python -c "python_code"
```

therefore the direction of effort is to write something like

```
TRANSFORM(...)
USING 'python -c "python_udf_code"'
```

The Python version on the Hive server machines in my environment is 2.7. The Hive version is 0.13, which is five years old as of this writing. (This will come haunt me in just a little bit. Keep reading.)

First, let's make up a naive UDF to see it *can* work.
The most naive UDF I can think of contains one statement, `print 123`:

```
hive> SELECT TRANSFORM(id, info_json) USING 'python -c "print 123"' AS (out STRING) FROM (SELECT * FROM cars) AS tmp;
Total jobs = 1
Launching Job 1 out of 1
... ...
Total MapReduce CPU Time Spent: 1 seconds 900 msec
OK
123
Time taken: 303.91 seconds, Fetched: 1 row(s)
hive>
```

Although painfully slow, the `123` in the print-out is really encouraging!

What if I want to print a string instead of the number `123`? The quotes around the string need to be escaped, which shouldn't be much of an issue. Let's try

```
hive> SELECT TRANSFORM(id, info_json) USING 'python -c "print \'abc\'"' AS (out STRING) FROM (SELECT * FROM cars) AS tmp;
...
Ended Job = job_1558314961036_2285 with errors
Error during job, obtaining debugging information...
...
Cased by: org.apache.hadoop.hive.ql.metadata.HiveException: [Error 20003]: An error occurred when trying to close the Operator running your custom script.
...
FAILED: Execution Error, returning code 20003 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask. An error occurred when trying to close the Operator running your custom script.
...

Ooops! The straightforward quote escaping does not work.
Let's skip this for now.
The above is running the Hive CLI on a "gateway" box.
Let's switch to my laptop and doing tests using our Hive client utility mentioned above.
{'fraction_non_alphanumeric': 0.0919607190412783, 'fraction_numerical': 0.012649800266311585, 'mean_word_length': 3.7441768653770233, 'pattern_counts': {'":': 22, '<': 5, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2462599', 'n_tokens_mistral': 3869, 'n_tokens_neox': 3597, 'n_words': 1929}