airflow/contrib/operators/sqoop_operator.py
def execute(self, context):
        """
        Execute sqoop job
        """
        self.hook = SqoopHook(
            conn_id=self.conn_id,
            verbose=self.verbose,
            num_mappers=self.num_mappers,
            hcatalog_database=self.hcatalog_database,
            hcatalog_table=self.hcatalog_table,
            properties=self.properties
        )

        if self.cmd_type == 'export':
            self.hook.export_table(
                table=self.table,
                export_dir=self.export_dir,
                input_null_string=self.input_null_string,
                input_null_non_string=self.input_null_non_string,
                staging_table=self.staging_table,
                clear_staging_table=self.clear_staging_table,
                enclosed_by=self.enclosed_by,
                escaped_by=self.escaped_by,
                input_fields_terminated_by=self.input_fields_terminated_by,
                input_lines_terminated_by=self.input_lines_terminated_by,
                input_optionally_enclosed_by=self.input_optionally_enclosed_by,
                batch=self.batch,
                relaxed_isolation=self.relaxed_isolation,
                extra_export_options=self.extra_export_options)
        elif self.cmd_type == 'import':
            # add create hcatalog table to extra import options if option passed
            # if new params are added to constructor can pass them in here
            # so don't modify sqoop_hook for each param
            if self.create_hcatalog_table:
                self.extra_import_options['create-hcatalog-table'] = ''

            if self.table and self.query:
                raise AirflowException(
                    'Cannot specify query and table together. Need to specify either or.'
                )

            if self.table:
                self.hook.import_table(
                    table=self.table,
                    target_dir=self.target_dir,
                    append=self.append,
                    file_type=self.file_type,
                    columns=self.columns,
                    split_by=self.split_by,
                    where=self.where,
                    direct=self.direct,
                    driver=self.driver,
                    extra_import_options=self.extra_import_options)
            elif self.query:
                self.hook.import_query(
                    query=self.query,
                    target_dir=self.target_dir,
                    append=self.append,
                    file_type=self.file_type,
                    split_by=self.split_by,
                    direct=self.direct,
                    driver=self.driver,
                    extra_import_options=self.extra_import_options)
            else:
                raise AirflowException(
                    "Provide query or table parameter to import using Sqoop"
                )
        else:
            raise AirflowException("cmd_type should be 'import' or 'export'")
def execute(self, context):
        """
        Execute sqoop job
        """
        self.hook = SqoopHook(
            conn_id=self.conn_id,
            verbose=self.verbose,
            num_mappers=self.num_mappers,
            hcatalog_database=self.hcatalog_database,
            hcatalog_table=self.hcatalog_table,
            properties=self.properties
        )

        if self.cmd_type == 'export':
            self.hook.export_table(
                table=self.table,
                export_dir=self.export_dir,
                input_null_string=self.input_null_string,
                input_null_non_string=self.input_null_non_string,
                staging_table=self.staging_table,
                clear_staging_table=self.clear_staging_table,
                enclosed_by=self.enclosed_by,
                escaped_by=self.escaped_by,
                input_fields_terminated_by=self.input_fields_terminated_by,
                input_lines_terminated_by=self.input_lines_terminated_by,
                input_optionally_enclosed_by=self.input_optionally_enclosed_by,
                batch=self.batch,
                relaxed_isolation=self.relaxed_isolation,
                extra_export_options=self.extra_export_options)
        elif self.cmd_type == 'import':
            # add create hcatalog table to extra import options if option passed
            # if new params are added to constructor can pass them in here
            # so don't modify sqoop_hook for each param
            if self.create_hcatalog_table:
                self.extra_import_options['create-hcatalog-table'] = ''

            if self.table and self.query:
                raise AirflowException(
                    'Cannot specify query and table together. Need to specify either or.'
                )

            if self.table:
                self.hook.import_table(
                    table=self.table,
                    target_dir=self.target_dir,
                    append=self.append,
                    file_type=self.file_type,
                    columns=self.columns,
                    split_by=self.split_by,
                    where=self.where,
                    direct=self.direct,
                    driver=self.driver,
                    extra_import_options=self.extra_import_options)
            elif self.query:
                self.hook.import_query(
                    query=self.query,
                    target_dir=self.target_dir,
                    append=self.append,
                    file_type=self.file_type,
                    split_by=self.split_by,
                    direct=self.direct,
                    driver=self.driver,
                    extra_import_options=self.extra_import_options)
            else:
                raise AirflowException(
                    "Provide query or table parameter to import using Sqoop"
                )
        else:
            raise AirflowException("cmd_type should be 'import' or 'export'")
def execute ( self , context ) : self . hook = SqoopHook ( conn_id = self . conn_id , verbose = self . verbose , num_mappers = self . num_mappers , hcatalog_database = self . hcatalog_database , hcatalog_table = self . hcatalog_table , properties = self . properties ) if self . cmd_type == 'export' : self . hook . export_table ( table = self . table , export_dir = self . export_dir , input_null_string = self . input_null_string , input_null_non_string = self . input_null_non_string , staging_table = self . staging_table , clear_staging_table = self . clear_staging_table , enclosed_by = self . enclosed_by , escaped_by = self . escaped_by , input_fields_terminated_by = self . input_fields_terminated_by , input_lines_terminated_by = self . input_lines_terminated_by , input_optionally_enclosed_by = self . input_optionally_enclosed_by , batch = self . batch , relaxed_isolation = self . relaxed_isolation , extra_export_options = self . extra_export_options ) elif self . cmd_type == 'import' : # add create hcatalog table to extra import options if option passed # if new params are added to constructor can pass them in here # so don't modify sqoop_hook for each param if self . create_hcatalog_table : self . extra_import_options [ 'create-hcatalog-table' ] = '' if self . table and self . query : raise AirflowException ( 'Cannot specify query and table together. Need to specify either or.' ) if self . table : self . hook . import_table ( table = self . table , target_dir = self . target_dir , append = self . append , file_type = self . file_type , columns = self . columns , split_by = self . split_by , where = self . where , direct = self . direct , driver = self . driver , extra_import_options = self . extra_import_options ) elif self . query : self . hook . import_query ( query = self . query , target_dir = self . target_dir , append = self . append , file_type = self . file_type , split_by = self . split_by , direct = self . direct , driver = self . driver , extra_import_options = self . extra_import_options ) else : raise AirflowException ( "Provide query or table parameter to import using Sqoop" ) else : raise AirflowException ( "cmd_type should be 'import' or 'export'" )
Execute sqoop job
Execute sqoop job