· 10 years ago · Sep 23, 2016, 01:36 PM
1import com.facebook.dataflow.DataFlow
2import com.facebook.dataflow.operators._
3
4val createTable = new MySQLOperator(
5 depList=List(),
6 mysqlTier="xdb.diet_query_analysis",
7 mysqlQuery=
8 """
9 |CREATE TABLE IF NOT EXISTS `<TABLE:hive_queries_from_log>` (
10 | `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
11 | `ds` VARCHAR(20) NOT NULL,
12 | `namespace_name` VARCHAR(50) NOT NULL,
13 | `task_name` VARCHAR(500) NOT NULL,
14 | `type` VARCHAR(20) NOT NULL,
15 | `query` LONGTEXT NOT NULL,
16 | `signature` CHAR(32) NOT NULL,
17 | `query_stats` TEXT,
18 | PRIMARY KEY (
19 | `id`
20 | ),
21 | KEY `query_sig` (
22 | `ds`,
23 | `signature`
24 | ),
25 | KEY `task_index` (
26 | `ds`,
27 | `task_name`
28 | ),
29 | KEY `namespace_index` (
30 | `ds`,
31 | `namespace_name`
32 | )
33 |) ENGINE=InnoDB DEFAULT CHARSET=latin1;
34 """.stripMargin
35)
36
37val waitLogPartition = new WaitForHiveOperator(
38 table="hive_command_log_signal",
39 partition="ds=<DATEID>"
40)
41
42val waitStatsPartition = new WaitForHiveOperator(
43 table="hive_query_stats",
44 partition="ds=<DATEID>"
45)
46
47val scrapeHiveQuery = new Hive2MySQLOperator(
48 depList=List(createTable, waitLogPartition, waitStatsPartition),
49 mysqlTier="xdb.diet_query_analysis",
50 mysqlTable="<TABLE:hive_queries_from_log>",
51 replica=false,
52 hiveQuery=
53 """
54 |SELECT
55 | NULL,
56 | log.ds,
57 | FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.namespace') AS ns,
58 | CONCAT(
59 | FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.pipelineName'),
60 | '.',
61 | FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.taskName')
62 | ) AS taskName,
63 | FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.type') AS type,
64 | log.command,
65 | FB_MD5(log.command),
66 | FB_MAKE_JSON_OBJ(MAP(
67 | 'inputSize', stats.input_size,
68 | 'outputSize', stats.output_size,
69 | 'taskRuntimeSec', stats.runtime_total_s,
70 | 'mappers', stats.mappers,
71 | 'reducers', stats.reducers,
72 | 'cpuSec', stats.cpu_msecs / 1000,
73 | 'mapCpuSec', stats.map_cpu_msecs / 1000,
74 | 'reduceCpuSec', stats.reduce_cpu_msecs / 1000,
75 | 'mapRuntimeSec', stats.map_runtime_total_s,
76 | 'reduceRuntimeSec', stats.reduce_runtime_total_s)
77 | ) AS query_stats
78 |FROM hive_command_log log
79 |LEFT OUTER JOIN (
80 | SELECT
81 | queryid,
82 | input_size,
83 | output_size,
84 | job_task_runtime_total_s AS runtime_total_s,
85 | mappers,
86 | reducers,
87 | cpu_msecs,
88 | map_cpu_msecs,
89 | reduce_cpu_msecs,
90 | map_task_runtime_total_s as map_runtime_total_s,
91 | reduce_task_runtime_total_s as reduce_runtime_total_s
92 | FROM hive_query_stats
93 | WHERE ds = '<DATEID>'
94 |) stats
95 |ON log.queryid = stats.queryid AND
96 |FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.taskName') IS NOT NULL
97 |WHERE log.command_type='QUERY'
98 |AND (
99 | FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.type') = 'DATASWARM'
100 | OR FB_JSON_PATH_EXTRACTOR(log.hive_query_source_json, '$.type') = 'DATABEE'
101 |) AND log.ds='<DATEID>'
102 """.stripMargin
103)
104
105val parseHiveQuery = new BashOperator(
106 depList=List(scrapeHiveQuery),
107 bashScript="/usr/local/jdk-8u60-64/bin/java " +
108 "-Xms512m -Xmx4g " +
109 "-cp <FBPACKAGE:constance/analyzer>/" +
110 "constance-analyzer-0.1.0-SNAPSHOT-jar-with-dependencies.jar " +
111 "-Djava.net.preferIPv6Addresses=true " +
112 "-Dconstance.root.logger=<TMP_DIR:constance/logs> " +
113 "-Dconstance.service.thread-pool.size=40 " +
114 "-Dconstance.hive-storage.queries-features-table-name=" +
115 "<TABLE:hive_query_features> " +
116 "com.facebook.constance.cli.AnalyzerCLI -pq " +
117 "-ds <DATEID>"
118)
119
120val createHiveTable = new HiveQLOperator(
121 depList=List(parseHiveQuery),
122 hiveQuery=
123 """
124 |CREATE TABLE IF NOT EXISTS <TABLE:fct_hive_queries> (
125 | id BIGINT,
126 | namespace_name STRING,
127 | task_name STRING,
128 | pipeline_type STRING,
129 | query STRING,
130 | query_signature STRING,
131 | query_ast STRING,
132 | query_stats STRING,
133 | query_features STRING,
134 | parse_failed INT
135 |) PARTITIONED BY (ds STRING)
136 | TBLPROPERTIES('RETENTION'='30');
137 """.stripMargin
138)
139
140val dumpToHive = new MySQL2HiveOperator(
141 depList=List(createHiveTable, parseHiveQuery),
142 mysqlTier="xdb.diet_query_analysis",
143 mysqlQuery=
144 """
145 |SELECT
146 | a.id,
147 | a.namespace_name,
148 | a.task_name,
149 | a.`type` as pipeline_type,
150 | a.query,
151 | a.signature as query_signature,
152 | b.parse_tree as query_ast,
153 | a.query_stats as query_stats,
154 | b.features as query_features,
155 | b.failed as parse_failed
156 |FROM hive_queries_from_log a
157 |LEFT JOIN hive_query_features b
158 |ON a.id = b.qid
159 |WHERE a.ds = '<DATEID>' AND b.ds = '<DATEID>';
160 """.stripMargin,
161 hiveTable="<TABLE:fct_hive_queries>",
162 hiveTartition="ds='<DATEID>'"
163)
164
165GlobalDefaults.set(
166 user="zjshen",
167 schedule="@daily",
168 numRetries=2,
169 dependsOnPast=false,
170 pool="di.lowpri_pipelines",
171 retryWait=300,
172 replica=false,
173 createTasks=true
174)
175
176DataFlow.execute(dumpToHive)