Luigi: Hive Examples Needed

[ericjmiller]

I'm having difficulty creating a trivial workflow script. I simply want to create a table (if it's not already there) viz CreateTable and put query results into that table viz HiveQuery. Here is what I've writting so far:

import luigi
import luigi.contrib.hive


class CreateTable(luigi.contrib.hive.HiveQueryTask):
    def query(self):
        return 'create table temp_batting_agg (col_value STRING);'


class HiveQuery(luigi.contrib.hive.HiveQueryTask):
    def requires(self):
        return CreateTable()

    def query(self):
        return 'select * from temp_batting limit 5;'

    def output(self):
        return luigi.contrib.hive.HiveTableTarget('temp_batting_agg')

if __name__ == "__main__":
    luigi.run()

The current output:

[root@sandbox luigi]# PYTHONPATH=. luigi --module batting_flow HiveQuery --local-scheduler
DEBUG: Checking if HiveQuery() is complete
DEBUG: Checking Hive table 'default.temp_batting_agg' exists
DEBUG: Checking if CreateTable() is complete
/usr/local/lib/python3.5/site-packages/luigi/worker.py:285: UserWarning: Task CreateTable() without outputs has no cus
tom complete() method
  is_complete = task.complete()
INFO: Informed scheduler that task   HiveQuery__99914b932b   has status   PENDING
INFO: Informed scheduler that task   CreateTable__99914b932b   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 16153] Worker Worker(salt=567817200, workers=1, host=sandbox.hortonworks.com, username=root, pid=16153) run
ning   CreateTable()
INFO: ['hive', '-f', '/tmp/tmp9_h06j9n', '--hiveconf', "mapred.job.name='CreateTable__99914b932b'"]
INFO: hive -f /tmp/tmp9_h06j9n --hiveconf mapred.job.name='CreateTable__99914b932b'
INFO: WARNING: Use "yarn jar" to launch YARN applications.
INFO: Logging initialized using configuration in file:/etc/hive/2.4.0.0-169/0/hive-log4j.properties
INFO: OK
INFO: Time taken: 7.972 seconds
INFO: [pid 16153] Worker Worker(salt=567817200, workers=1, host=sandbox.hortonworks.com, username=root, pid=16153) done      CreateTable()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   CreateTable__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 16153] Worker Worker(salt=567817200, workers=1, host=sandbox.hortonworks.com, username=root, pid=16153) running   HiveQuery()
/usr/local/lib/python3.5/site-packages/luigi/worker.py:156: UserWarning: Task CreateTable() without outputs has no custom complete() method
  missing = [dep.task_id for dep in self.task.deps() if not dep.complete()]
ERROR: [pid 16153] Worker Worker(salt=567817200, workers=1, host=sandbox.hortonworks.com, username=root, pid=16153) failed    HiveQuery()
Traceback (most recent call last):
  File "/usr/local/lib/python3.5/site-packages/luigi/worker.py", line 159, in run
    raise RuntimeError('Unfulfilled %s at run time: %s' % (deps, ', '.join(missing)))
RuntimeError: Unfulfilled dependency at run time: CreateTable__99914b932b
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Skipping error email. Set `error-email` in the `core` section of the luigi config file or override `owner_email`in the task to receive error emails.
INFO: Informed scheduler that task   HiveQuery__99914b932b   has status   FAILED
DEBUG: Checking if HiveQuery() is complete
DEBUG: Checking Hive table 'default.temp_batting_agg' exists
WARNING: Will not schedule HiveQuery() or any dependencies due to error in complete() method:
Traceback (most recent call last):
  File "/usr/local/lib/python3.5/site-packages/luigi/worker.py", line 285, in check_complete
    is_complete = task.complete()
  File "/usr/local/lib/python3.5/site-packages/luigi/task.py", line 389, in complete
    return all(map(lambda output: output.exists(), outputs))
  File "/usr/local/lib/python3.5/site-packages/luigi/task.py", line 389, in <lambda>
    return all(map(lambda output: output.exists(), outputs))
  File "/usr/local/lib/python3.5/site-packages/luigi/contrib/hive.py", line 382, in exists
    return self.client.table_exists(self.table, self.database)
  File "/usr/local/lib/python3.5/site-packages/luigi/contrib/hive.py", line 141, in table_exists
    return stdout and table.lower() in stdout
TypeError: a bytes-like object is required, not 'str'

INFO: Skipping error email. Set `error-email` in the `core` section of the luigi config file or override `owner_email`in the task to receive error emails.
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=567817200, workers=1, host=sandbox.hortonworks.com, username=root, pid=16153) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 2 tasks of which:
* 1 ran successfully:
    - 1 CreateTable(pool=None)
* 1 failed:
    - 1 HiveQuery(pool=None)

This progress looks :( because there were failed tasks

===== Luigi Execution Summary =====

I'll submit a PR with a solid Hive example - I think it's needed because I'm having significant trouble doing anything with the results of a query. The query will run, but I can't manipulate them or write them to a table, etc.

Thanks!

[impguard]

Here's an example of something we've spun up internally to handle running hive queries. In fact, we this similar model for running any sort of query that uses subprocess. We didn't like the model presented in the contrib hive (global hive cli config etc.) and used this instead.

class HiveQuery(luigi.Task):
    """ Runs a hive query.

    A helpful task that encapsulates some useful logic when running a hive
    query and captures the necessary parameters to do so. Provides a helper
    method `execute` to run the hive command specified that returns a
    subprocess.CompletedProcess:
    https://docs.python.org/3/library/subprocess.html#subprocess.CompletedProcess

    By default, execute opens up stdout and stderr so the CompletedProcess
    will contain stdout and stderr.

    By default, execution will error if the hive command exists with a non-zero
    exit code.
    """
    hive_cli          = luigi.Parameter(default='beeline')

    hive_driver       = luigi.Parameter(default='jdbc:hive2')
    hive_server       = luigi.Parameter()
    hive_port         = luigi.Parameter()
    hive_database     = luigi.Parameter()

    hive_username     = luigi.Parameter(significant=False)
    hive_password     = luigi.Parameter(significant=False)

    @property
    def hive_query(self):
        raise NotImplementedError()

    @property
    def hive_args(self):
        """ Additional hive arguments if necessary."""
        return []

    @property
    def hive_base_command(self):
        server_path = self.hive_server + ':' + self.hive_port
        url_parts = (self.hive_driver, server_path, self.hive_database, '', '')
        connection_string = urlunsplit(url_parts)

        base_hive_command = [
            self.hive_cli,
            '-u', connection_string,
            '-n', self.hive_username,
            '-p', self.hive_password]

        return base_hive_command

    @property
    def hive_command(self):
        if not self.hive_query:
            raise ValueError('A hive query needs to be provided to be run.')

        hive_command = (
            self.hive_base_command +
            ['-e', self.hive_query] +
            self.hive_args)

        return hive_command

    def execute(self):
        log.debug('Executing query: {}'.format(self.hive_query))
        log.debug('Additional query args: {}'.format(self.hive_args))

        result = subprocess.run(
            self.hive_command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=True)

        # TODO: Figure out a way to check for error
        return result

    def run(self):
        raise NotImplementedError()

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.