mysql数据倾斜_sqoop数据倾斜解决实战

背景: 下面导致数据倾斜的sqoop任务,这个数据倾斜是由于源数据导致的- 源表的index1 分布不均匀,不均匀指每一块的数据量不均匀&＃xff0c;范围[min(index),max(index)]被分为m块&＃xff0c;每块index[0,7]的值在 [min&＃43;index*range/8,min&＃43;(index&＃43;1)*range] 间&＃xff0c;如第一块100MB,最后一块1GB的数据量&＃xff0c;数据量可在hdfs查看map文件大小。

(数据量&＃xff0c;用时)&＃xff1a;(一千万&＃43;条&＃xff0c;20分钟)

sqoop import \

--connect "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}?zeroDateTimeBehavior&＃61;convertToNull&dontTrackOpenResources&＃61;true&tinyInt1isBit&＃61;false&defaultFetchSize&＃61;10000&useCursorFetch&＃61;true&autoReconnect&＃61;true&failOverReadOnly&＃61;false&useSSL&＃61;false" \

--driver com.mysql.jdbc.Driver \

--username ${mysql_user} \

--password ${mysql_password} \

--table ${mysql_table} \

--columns &＃39;field1,field2,...,fieldn&＃39; \

--hcatalog-database test \

--hcatalog-table ${hive_table} \

--hcatalog-partition-keys 分区字段 \

--hcatalog-partition-values 分区字段值 \

-m 8 \

--split-by index1

solutionA: 使用–query语句中添加自增ID&＃xff0c;作为split-by的参数

用时&＃xff1a;8分钟

sqoop import \

--connect "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}?zeroDateTimeBehavior&＃61;convertToNull&dontTrackOpenResources&＃61;true&tinyInt1isBit&＃61;false&defaultFetchSize&＃61;10000&useCursorFetch&＃61;true&autoReconnect&＃61;true&failOverReadOnly&＃61;false&useSSL&＃61;false" \

--driver com.mysql.jdbc.Driver \

--username ${mysql_user} \

--password ${mysql_password} \

--hcatalog-database test \

--hcatalog-table table \

--hcatalog-partition-keys 分区字段 \

--hcatalog-partition-values 分区字段值 \

--query &＃39;select tt.* from (select &＃64;rowNum:&＃61;&＃64;rowNum&＃43;1 as INC_ID, t.* from table t,(select &＃64;rowNum:&＃61;0) b ) tt where $CONDITIONS&＃39; \

-boundary-query "select 1 as min , sum(1) as max from table " \

-m 8 \

--split-by INC_ID

solutionB: 使用另一个均匀索引-index2

用时&＃xff1a;8分钟

sqoop import \

--connect "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}?zeroDateTimeBehavior&＃61;convertToNull&dontTrackOpenResources&＃61;true&tinyInt1isBit&＃61;false&defaultFetchSize&＃61;10000&useCursorFetch&＃61;true&autoReconnect&＃61;true&failOverReadOnly&＃61;false&useSSL&＃61;false" \

--driver com.mysql.jdbc.Driver \

--username ${mysql_user} \

--password ${mysql_password} \

--table ${mysql_table} \

--columns &＃39;field1,field2,...,fieldn&＃39; \

--hcatalog-database test \

--hcatalog-table ${hive_table} \

--hcatalog-partition-keys 分区字段 \

--hcatalog-partition-values 分区字段值 \

-m 8 \

--split-by index2

solutionB_update: 查看了planB的执行计划&＃xff0c;发现是全表扫描&＃xff0c;要通过force index(index) 强制执行索引扫描

用时&＃xff1a;3分钟

sqoop import \

--connect --connect "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}?zeroDateTimeBehavior&＃61;convertToNull&dontTrackOpenResources&＃61;true&tinyInt1isBit&＃61;false&defaultFetchSize&＃61;10000&useCursorFetch&＃61;true&autoReconnect&＃61;true&failOverReadOnly&＃61;false&useSSL&＃61;false" \

--driver com.mysql.jdbc.Driver \

--username ${mysql_user} \

--password ${mysql_password} \

--hcatalog-database test \

--hcatalog-table table \

--hcatalog-partition-keys 分区字段 \

--hcatalog-partition-values 分区字段值 \

--query &＃39;select field1,...,fieldn from table T force index(index2) where $CONDITIONS&＃39; \

-m 8 \

--split-by index2

推荐排名&＃xff1a;solutionB_update > solutionB > solutionA

适用场景&＃xff1a;solutionB_update/solutionB 适用于mysql有均匀索引的情况。没有均匀索引的话&＃xff0c;用solutionA。由于一天的数据量很大(1000w&＃43;)&＃xff0c;应该考虑增量更新的方式-canal 导入&＃xff0c;但听说使用canal会比较麻烦&＃xff0c;有待研究,目前先用这些方案。

sqoop参数说明(来源官网)&＃xff1a;

翻译&＃xff1a;

7.2.2。选择要导入的数据

Sqoop通常以表为中心导入数据。使用--table参数选择要导入的表。例如&＃xff0c;--table employees。此参数还可以标识数据库中的视图或其他类似表的实体。

默认情况下&＃xff0c;表中的所有列都被选中进行导入。导入的数据以其“自然顺序”写入HDFS&＃xff1b;也就是说&＃xff0c;包含列a、B和C的表会导致数据导入&＃xff0c;例如&＃xff1a;

A1、B1、C1

A2、B2、C2

...

您可以选择列的子集并使用--columns参数控制它们的顺序。这应该包括要导入的列的逗号分隔列表。例如&＃xff1a;--列“name&＃xff0c;employee_id&＃xff0c;jobtitle”。

可以通过向import语句中添加sqlwhere子句来控制导入哪些行。默认情况下&＃xff0c;Sqoop生成SELECTFROM

默认情况下&＃xff0c;sqoop将使用query select min()&＃xff0c;max()from

标签&＃xff1a;实战,倾斜,hcatalog,--,sqoop,mysql,table,false,true

来源&＃xff1a; https://blog.csdn.net/AnlaGodness/article/details/110090691