<s> import logging <EOL> from airflow . hooks import HiveCliHook , DruidHook , HiveMetastoreHook <EOL> from airflow . models import BaseOperator <EOL> from airflow . utils . decorators import apply_defaults <EOL> class HiveToDruidTransfer ( BaseOperator ) : <EOL> """<STR_LIT>""" <EOL> template_fields = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> template_ext = ( '<STR_LIT>' , ) <EOL> @ apply_defaults <EOL> def __init__ ( <EOL> self , <EOL> sql , <EOL> druid_datasource , <EOL> ts_dim , <EOL> metric_spec = None , <EOL> hive_cli_conn_id = '<STR_LIT>' , <EOL> druid_ingest_conn_id = '<STR_LIT>' , <EOL> metastore_conn_id = '<STR_LIT>' , <EOL> hadoop_dependency_coordinates = None , <EOL> intervals = None , <EOL> num_shards = <NUM_LIT:1> , <EOL> * args , ** kwargs ) : <EOL> super ( HiveToDruidTransfer , self ) . __init__ ( * args , ** kwargs ) <EOL> self . sql = sql <EOL> self . druid_datasource = druid_datasource <EOL> self . ts_dim = ts_dim <EOL> self . intervals = intervals or [ '<STR_LIT>' ] <EOL> self . num_shards = num_shards <EOL> self . metric_spec = metric_spec or [ { <EOL> "<STR_LIT:name>" : "<STR_LIT:count>" , <EOL> "<STR_LIT:type>" : "<STR_LIT:count>" } ] <EOL> self . hive_cli_conn_id = hive_cli_conn_id <EOL> self . hadoop_dependency_coordinates = hadoop_dependency_coordinates <EOL> self . druid_ingest_conn_id = druid_ingest_conn_id <EOL> self . metastore_conn_id = metastore_conn_id <EOL> def execute ( self , context ) : <EOL> hive = HiveCliHook ( hive_cli_conn_id = self . hive_cli_conn_id ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> hive_table = '<STR_LIT>' + context [ '<STR_LIT>' ] . replace ( '<STR_LIT:.>' , '<STR_LIT:_>' ) <EOL> sql = self . sql . strip ( ) . strip ( '<STR_LIT:;>' ) <EOL> hql = """<STR_LIT>""" . format ( ** locals ( ) ) <EOL> logging . info ( "<STR_LIT>" . format ( hql ) ) <EOL> hive . run_cli ( hql ) <EOL> m = HiveMetastoreHook ( self . metastore_conn_id ) <EOL> t = m . get_table ( hive_table ) <EOL> columns = [ col . name for col in t . sd . cols ] <EOL> hdfs_uri = m . get_table ( hive_table ) . sd . location <EOL> pos = hdfs_uri . find ( '<STR_LIT>' ) <EOL> static_path = hdfs_uri [ pos : ] <EOL> schema , table = hive_table . split ( '<STR_LIT:.>' ) <EOL> druid = DruidHook ( druid_ingest_conn_id = self . druid_ingest_conn_id ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" + static_path ) <EOL> druid . load_from_hdfs ( <EOL> datasource = self . druid_datasource , <EOL> intervals = self . intervals , <EOL> static_path = static_path , ts_dim = self . ts_dim , <EOL> columns = columns , num_shards = self . num_shards , metric_spec = self . metric_spec , <EOL> hadoop_dependency_coordinates = self . hadoop_dependency_coordinates ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( hive_table ) ) <EOL> hql = "<STR_LIT>" . format ( hive_table ) <EOL> hive . run_cli ( hql ) </s>
