最近一次數據遷移,須要將MySQL的數據導出、處理後導入到新表和ES。這裏作個簡單記錄,方便後續查詢。php
注: 爲了寫文章方便及隱私安全,實際內容會有所簡化。例如表結構簡化、數據庫鏈接部分所有用 xxx 表示、目錄及文件名均爲化名等。html
原表:mysql
book_db 庫 - b_book(id,create_time,update_time,price,title,intro)
新表:sql
book 庫 - book(id,price,title,create_time,update_time) - book_ext(id,book_id,intro,create_time)
mkdir -p /tmp/ # 導出原始數據 mysql -hxxx -uxxx -pxxx book_db --default-character-set=utf8 -e 'select id,create_time,update_time,price,title,intro from b_book' | sed 's/NULL//g' > /tmp/b_book.csv
sed 's/NULL//g'
是由於導出的數據有些字段存的NULL,新表不須要存儲NULL,因此去掉。shell
導出的數據每行默認以\t
分隔,第一行包含字段名。這裏咱們刪掉第一行:數據庫
sed -i '1d' /tmp/b_book.csv
cd /tmp/ # 處理create_time,update_time,price,並生成文件 book.csv cat b_book.csv | awk -F '\t' -v OFS=' @@@ ' '{gsub(/[-:]/," ",$2); $2=mktime($2);gsub(/[-:]/,"",$3);$3=mktime($3);$4=$4*100;$6="";print $0}' > book.csv # 生成文件 book_ext.csv cat b_book.csv | awk -F '\t' -v OFS=' @@@ ' '{print $1,$6}' > book_ext.csv # 生成文件 book_es.csv cat b_book.csv | awk -F '\t' -v OFS=' @@@ ' '{$4=$4*100;print $0}' > book_es.csv
由於原表裏時間都是datetime格式,新表是時間戳格式,這裏處理成時間戳格式。價格原表是以元爲單位,這裏*100
是爲了處理成以分爲單位。json
-v OFS=' @@@ '
表示輸出的時候每列以@@@
爲分隔符。緣由是原表裏的intro
字段存儲的是html,可能包含經常使用轉義字符,這裏使用@@@
確保能正確分隔每列。安全
mysql -hxxx -uxxx -pxxx book
Load Data LOCAL InFile '/tmp/book.csv' Into Table book character set utf8 Fields Terminated By ' @@@ ' Enclosed By '' Escaped By '' Lines Terminated By '\n' (id,create_time,update_time,price,title); Load Data LOCAL InFile '/tmp/book_ext.csv' Into Table book_ext character set utf8 Fields Terminated By ' @@@ ' Enclosed By '' Escaped By '' Lines Terminated By '\n' (book_id,intro);
說明:bash
\t
Into Table
表明插入,記錄已存在(惟一鍵約束)則失敗再也不往下執行。Replace Into Table
表明覆蓋,記錄已存在則覆蓋(是整條記錄覆蓋,沒有列出的字段給默認值)。Ignore Into Table
遇到已存在直接跳過。app
因爲生產的book_es.csv
文件比較大,因此這裏按20000條生成一個文件,防止文件過大,ES導入失敗。
cd /tmp/ awk '{filename = "book_es.csv." int((NR-1)/20000) ".csv"; print >> filename}' book_es.csv
ConvertBookToEs.php
是PHP腳本,生成ES批量導入的文件。見附錄。執行後生成不少book_es.csv.*.csv.json
文件。
php ConvertBookToEs.php
importToEs.sh
是ES批量導入腳本,以下:
#!/bin/bash for file in `ls /tmp/book_es.csv.*.csv.json` do echo $file; curl -XPOST http://xxx:9200/book/doc/_bulk -H "Content-Type: application/json" --data-binary "@$file" >> importToEs.log done
執行腳本:
sh importToEs.sh
等待數分鐘,便執行完畢了。
爲了將大量數據加載到MySQL中,LOAD DATA INFILE
是迄今爲止最快的選擇。可是,雖然這能夠以INSERT IGNORE
或REPLACE
的方式使用,但目前不支持ON DUPLICATE KEY UPDATE
。
若是咱們想批量更新某個字段,ON DUPLICATE KEY UPDATE
如何使用LOAD DATA INFILE
模擬?
stackoverflow 上有網友給了答案。步驟是:
1)建立一個新的臨時表。
CREATE TEMPORARY TABLE temporary_table LIKE target_table;
2)從臨時表中刪除全部索引以加快速度。(可選)
SHOW INDEX FROM temporary_table; DROP INDEX `PRIMARY` ON temporary_table; DROP INDEX `some_other_index` ON temporary_table;
3)將CSV加載到臨時表中
LOAD DATA INFILE 'your_file.csv' INTO TABLE temporary_table Fields Terminated By '\t' Enclosed By '' Escaped By '' Lines Terminated By '\n' (field1, field2);
4)使用ON DUPLICATE KEY UPDATE
複製數據
SHOW COLUMNS FROM target_table; INSERT INTO target_table SELECT * FROM temporary_table ON DUPLICATE KEY UPDATE field1 = VALUES(field1), field2 = VALUES(field2);
MySQL將假定=
以前的部分引用INSERT INTO
子句中指定的列,第二部分引用SELECT
列。
5)刪除臨時表
DROP TEMPORARY TABLE temporary_table;
使用SHOW INDEX FROM
和SHOW COLUMNS FROM
此過程能夠針對任何給定的表自動執行。
注:官方文檔裏
INSERT ... SELECT ON DUPLICATE KEY UPDATE
語句被標記爲基於語句的複製不安全。因此上述方案請在充分測試後再實施。詳見:
https://dev.mysql.com/doc/refman/5.6/en/insert-on-duplicate.html
ConvertBookToEs.php
<?php /** * 轉換wish_book爲ES 批量格式(json) */ //id,create_time,update_time,price,title,intro function dealBook($file) { $fp = fopen($file, 'r'); while (!feof($fp)) { $line = explode(' @@@ ', fgets($fp, 65535)); if ($line && isset($line[1])) { $arr_head = [ 'index' => [ '_id' => (int)$line[0] ] ]; $arr = [ 'id' => (int)$line[0], 'create_time' => strtotime($line[1]), 'update_time' => strtotime($line[2]), 'price' => intval($line[3]), 'title' => (string)$line[4], 'intro' => (string)$line[18], ]; file_put_contents($file . '.json', json_encode($arr_head, JSON_UNESCAPED_UNICODE) . PHP_EOL, FILE_APPEND); file_put_contents($file . '.json', json_encode($arr, JSON_UNESCAPED_UNICODE) . PHP_EOL, FILE_APPEND); } } } try { //處理CSV文件爲es bluk json格式 //參考 https://www.elastic.co/guide/en/elasticsearch/reference/current/_batch_processing.html $files = glob("/tmp/book_es.csv.*.csv"); if (false === $files) { exit("can not find csv file"); } $pids = []; foreach ($files as $i => $file) { $pid = pcntl_fork(); if ($pid < 0) { exit("could not fork"); } if ($pid > 0) { $pids[$pid] = $pid; } else { echo time() . " new process, pid:" . getmypid() . PHP_EOL; dealBook($file); exit(); } } while (count($pids)) { foreach ($pids as $key => $pid) { $res = pcntl_waitpid($pid, $status, WNOHANG); if ($res == -1 || $res > 0) { echo 'Child process exit,pid ' . $pid . PHP_EOL; unset($pids[$key]); } } sleep(1); } } catch (Exception $e) { $message = $e->getFile() . ':' . $e->getLine() . ' ' . $e->getMessage(); echo $message; }
一、Linux命令行文本工具 - 飛鴻影~ - 博客園 https://www.cnblogs.com/52fhy/p/5836429.html 二、mysqldump 導出 csv 格式 --fields-terminated-by=, :字段分割符; - superhosts的專欄 - CSDN博客 https://blog.csdn.net/superhosts/article/details/26054997 三、Batch Processing | Elasticsearch Reference [6.4] | Elastic https://www.elastic.co/guide/en/elasticsearch/reference/current/_batch_processing.html 四、mysql導入數據load data infile用法整理 - conanwang - 博客園 https://www.cnblogs.com/conanwang/p/5890753.html 五、MySQL LOAD DATA INFILE with ON DUPLICATE KEY UPDATE - Stack Overflow https://stackoverflow.com/questions/15271202/mysql-load-data-infile-with-on-duplicate-key-update 六、mysql - INSERT INTO ... SELECT FROM ... ON DUPLICATE KEY UPDATE - Stack Overflow https://stackoverflow.com/questions/2472229/insert-into-select-from-on-duplicate-key-update 七、MySQL :: MySQL 5.6參考手冊:: 13.2.5.2 INSERT ... ON DUPLICATE KEY UPDATE語法 https://dev.mysql.com/doc/refman/5.6/en/insert-on-duplicate.html 八、複製表結構和數據SQL語句 - becket - 博客園 https://www.cnblogs.com/zhengxu/articles/2206894.html