airflow/hooks/hive_hooks.py
def load_file(
            self,
            filepath,
            table,
            delimiter=",",
            field_dict=None,
            create=True,
            overwrite=True,
            partition=None,
            recreate=False,
            tblproperties=None):
        """
        Loads a local file into Hive

        Note that the table generated in Hive uses ``STORED AS textfile``
        which isn't the most efficient serialization format. If a
        large amount of data is loaded and/or if the tables gets
        queried considerably, you may want to use this operator only to
        stage the data into a temporary table before loading it into its
        final destination using a ``HiveOperator``.

        :param filepath: local filepath of the file to load
        :type filepath: str
        :param table: target Hive table, use dot notation to target a
            specific database
        :type table: str
        :param delimiter: field delimiter in the file
        :type delimiter: str
        :param field_dict: A dictionary of the fields name in the file
            as keys and their Hive types as values.
            Note that it must be OrderedDict so as to keep columns' order.
        :type field_dict: collections.OrderedDict
        :param create: whether to create the table if it doesn't exist
        :type create: bool
        :param overwrite: whether to overwrite the data in table or partition
        :type overwrite: bool
        :param partition: target partition as a dict of partition columns
            and values
        :type partition: dict
        :param recreate: whether to drop and recreate the table at every
            execution
        :type recreate: bool
        :param tblproperties: TBLPROPERTIES of the hive table being created
        :type tblproperties: dict
        """
        hql = ''
        if recreate:
            hql += "DROP TABLE IF EXISTS {table};\n".format(table=table)
        if create or recreate:
            if field_dict is None:
                raise ValueError("Must provide a field dict when creating a table")
            fields = ",\n    ".join(
                [k + ' ' + v for k, v in field_dict.items()])
            hql += "CREATE TABLE IF NOT EXISTS {table} (\n{fields})\n".format(
                table=table, fields=fields)
            if partition:
                pfields = ",\n    ".join(
                    [p + " STRING" for p in partition])
                hql += "PARTITIONED BY ({pfields})\n".format(pfields=pfields)
            hql += "ROW FORMAT DELIMITED\n"
            hql += "FIELDS TERMINATED BY '{delimiter}'\n".format(delimiter=delimiter)
            hql += "STORED AS textfile\n"
            if tblproperties is not None:
                tprops = ", ".join(
                    ["'{0}'='{1}'".format(k, v) for k, v in tblproperties.items()])
                hql += "TBLPROPERTIES({tprops})\n".format(tprops=tprops)
        hql += ";"
        self.log.info(hql)
        self.run_cli(hql)
        hql = "LOAD DATA LOCAL INPATH '{filepath}' ".format(filepath=filepath)
        if overwrite:
            hql += "OVERWRITE "
        hql += "INTO TABLE {table} ".format(table=table)
        if partition:
            pvals = ", ".join(
                ["{0}='{1}'".format(k, v) for k, v in partition.items()])
            hql += "PARTITION ({pvals})".format(pvals=pvals)

        # As a workaround for HIVE-10541, add a newline character
        # at the end of hql (AIRFLOW-2412).
        hql += ';\n'

        self.log.info(hql)
        self.run_cli(hql)
def load_file(
            self,
            filepath,
            table,
            delimiter=",",
            field_dict=None,
            create=True,
            overwrite=True,
            partition=None,
            recreate=False,
            tblproperties=None):
        hql = ''
        if recreate:
            hql += "DROP TABLE IF EXISTS {table};\n".format(table=table)
        if create or recreate:
            if field_dict is None:
                raise ValueError("Must provide a field dict when creating a table")
            fields = ",\n    ".join(
                [k + ' ' + v for k, v in field_dict.items()])
            hql += "CREATE TABLE IF NOT EXISTS {table} (\n{fields})\n".format(
                table=table, fields=fields)
            if partition:
                pfields = ",\n    ".join(
                    [p + " STRING" for p in partition])
                hql += "PARTITIONED BY ({pfields})\n".format(pfields=pfields)
            hql += "ROW FORMAT DELIMITED\n"
            hql += "FIELDS TERMINATED BY '{delimiter}'\n".format(delimiter=delimiter)
            hql += "STORED AS textfile\n"
            if tblproperties is not None:
                tprops = ", ".join(
                    ["'{0}'='{1}'".format(k, v) for k, v in tblproperties.items()])
                hql += "TBLPROPERTIES({tprops})\n".format(tprops=tprops)
        hql += ";"
        self.log.info(hql)
        self.run_cli(hql)
        hql = "LOAD DATA LOCAL INPATH '{filepath}' ".format(filepath=filepath)
        if overwrite:
            hql += "OVERWRITE "
        hql += "INTO TABLE {table} ".format(table=table)
        if partition:
            pvals = ", ".join(
                ["{0}='{1}'".format(k, v) for k, v in partition.items()])
            hql += "PARTITION ({pvals})".format(pvals=pvals)

        # As a workaround for HIVE-10541, add a newline character
        # at the end of hql (AIRFLOW-2412).
        hql += ';\n'

        self.log.info(hql)
        self.run_cli(hql)
def load_file ( self , filepath , table , delimiter = "," , field_dict = None , create = True , overwrite = True , partition = None , recreate = False , tblproperties = None ) : hql = '' if recreate : hql += "DROP TABLE IF EXISTS {table};\n" . format ( table = table ) if create or recreate : if field_dict is None : raise ValueError ( "Must provide a field dict when creating a table" ) fields = ",\n    " . join ( [ k + ' ' + v for k , v in field_dict . items ( ) ] ) hql += "CREATE TABLE IF NOT EXISTS {table} (\n{fields})\n" . format ( table = table , fields = fields ) if partition : pfields = ",\n    " . join ( [ p + " STRING" for p in partition ] ) hql += "PARTITIONED BY ({pfields})\n" . format ( pfields = pfields ) hql += "ROW FORMAT DELIMITED\n" hql += "FIELDS TERMINATED BY '{delimiter}'\n" . format ( delimiter = delimiter ) hql += "STORED AS textfile\n" if tblproperties is not None : tprops = ", " . join ( [ "'{0}'='{1}'" . format ( k , v ) for k , v in tblproperties . items ( ) ] ) hql += "TBLPROPERTIES({tprops})\n" . format ( tprops = tprops ) hql += ";" self . log . info ( hql ) self . run_cli ( hql ) hql = "LOAD DATA LOCAL INPATH '{filepath}' " . format ( filepath = filepath ) if overwrite : hql += "OVERWRITE " hql += "INTO TABLE {table} " . format ( table = table ) if partition : pvals = ", " . join ( [ "{0}='{1}'" . format ( k , v ) for k , v in partition . items ( ) ] ) hql += "PARTITION ({pvals})" . format ( pvals = pvals ) # As a workaround for HIVE-10541, add a newline character # at the end of hql (AIRFLOW-2412). hql += ';\n' self . log . info ( hql ) self . run_cli ( hql )
Loads a local file into Hive

        Note that the table generated in Hive uses ``STORED AS textfile``
        which isn't the most efficient serialization format. If a
        large amount of data is loaded and/or if the tables gets
        queried considerably, you may want to use this operator only to
        stage the data into a temporary table before loading it into its
        final destination using a ``HiveOperator``.

        :param filepath: local filepath of the file to load
        :type filepath: str
        :param table: target Hive table, use dot notation to target a
            specific database
        :type table: str
        :param delimiter: field delimiter in the file
        :type delimiter: str
        :param field_dict: A dictionary of the fields name in the file
            as keys and their Hive types as values.
            Note that it must be OrderedDict so as to keep columns' order.
        :type field_dict: collections.OrderedDict
        :param create: whether to create the table if it doesn't exist
        :type create: bool
        :param overwrite: whether to overwrite the data in table or partition
        :type overwrite: bool
        :param partition: target partition as a dict of partition columns
            and values
        :type partition: dict
        :param recreate: whether to drop and recreate the table at every
            execution
        :type recreate: bool
        :param tblproperties: TBLPROPERTIES of the hive table being created
        :type tblproperties: dict
Loads a local file into Hive