因為做股票金融的，每天產生的資料量是很大的，一個月幾十億的交易記錄，也常有出現，特別是今年大跌之前大漲那會。

作為程式設計師，問題來了，有時需要將一些並不是特別符合規範的csv檔案匯入資料庫中，而且每個檔案有十幾萬行，而這樣的檔案幾萬個，於是幾十億的記錄如何匯入資料庫呢？很多想著常用的方法，就是將csv檔案讀出來，然後一條條插到資料庫，或者批量插到資料庫，或者開個多執行緒，然而，花費的時間必須是幾十天，而且不好控制，萬一中斷，那又不知該如何是好了。。。

對於我，我最先想到是開多執行緒，每次1000條批量插入資料庫，然而，由於我犯了一個致命的錯誤：對於大量資料修改，資料庫表先不要建索引，因為每次改動表，索引都會修改，這在資料量大的時候，消耗的時間也是比較大的。所以，我效果並不好，12個小時也才存了4千萬資料。看到這個效果，我感覺不對勁，因為我要存的資料，至少是幾十億的，於是我覺得肯定有更效率的方法，直到load data infile

才讓我覺得人生處處有驚喜！！

用LOAD DATA INFILE將1.csv匯入資料庫表d201505的語句如下：

LOAD DATA INFILE 'E:/1.csv'
INTO TABLE d201505
FIELDS TERMINATED BY ','
ENCLOSED BY '\"'
ESCAPED BY '\\'
LINES TERMINATED BY '\n'
(stockID,date,price,buysell,volume);

就這麼一條語句，一個含有十幾萬資料的匯入表中基本是秒級，於是我決定先將我的不規則的csv檔案讀寫成合乎資料表字段的csv，然後LOAD DATA INFILE匯入到資料表，並且我開了一百個多執行緒對檔案進行寫，開了三個執行緒對資料匯入，效果很樂觀，完整一個流程下來，10多億資料花了八個小時，平均十分鐘左右就能匯入1500萬資料，所以，在此記錄下，我的程式如下：

import java.io.File;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
//用了javacsv這個外掛包
import com.csvreader.CsvReader;
import com.csvreader.CsvWriter;

public class CSVReadAndWrite {
    // 寫cvs檔案
    public static void writeToCsv(String dirPath, String ym, String d,
            String[] cvsfile) {
        ExecutorService pool = Executors.newFixedThreadPool(100);
        for (int j = 0; j < cvsfile.length; j++) {
            Thread thread = new MDownThread(dirPath, cvsfile[j], ym, d);
            pool.execute(thread);

        }
        pool.shutdown();
        try {
            pool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    // 我匯入的是201504這一個月的資料，所以是1-30，按天匯入
    public static void main(String[] args) throws Exception {
        String ym = "201504";
        String d;

        for (int i = 1; i <= 30; i++) {
            Date t1 = new Date();
            String dirPath = "E:\\" + ym + "/" + ym;
            d = String.format("%02d", i);
            dirPath += d + "/";
            File file = new File(dirPath);
            String cvsfile[];
            cvsfile = file.list();
            if (cvsfile == null || cvsfile.length <= 0) {
                continue;
            }
            System.out.println(dirPath + "開始匯入！");
            // 寫入檔案
            writeToCsv(dirPath, ym, d, cvsfile);
            ExecutorService pool2 = Executors.newFixedThreadPool(3);
            for (int j = 0; j < cvsfile.length; j++) {
                if (!new File("E:/" + ym + "/" + ym + d + "/"
                        + cvsfile[j]).exists()) {
                    writeToCsv(dirPath, ym, d, cvsfile);
                }
                Thread thread2 = new LoadMysqlThread(dirPath, cvsfile[j], ym, d);
                pool2.execute(thread2);
            }
            pool2.shutdown();
            try {
                pool2.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            Date t2 = new Date();
            System.out.println(dirPath + "匯入成功,用時："
                    + (t2.getTime() - t1.getTime()) / 1000 + "s");

        }

    }

}
//執行緒寫csv檔案
class MDownThread extends Thread {
    SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    public String dirPath;
    public String cvsfile;
    public String ym;
    public String d;
    public int dj;

    public MDownThread(String dirPath, String cvsfile, String ym, String d) {
        this.dirPath = dirPath;
        this.cvsfile = cvsfile;
        this.ym = ym;
        this.d = d;
        this.dj = dj;
    }

    @Override
    public void run() {
        try {
            // 生成CsvReader物件，以，為分隔符，utf-8編碼方式
            CsvReader r = new CsvReader(dirPath + cvsfile, ',',
                    Charset.forName("utf-8"));
            CsvWriter wr = new CsvWriter("E:\\" + ym + "\\"
                    + ym + d + "\\" + cvsfile, ',', Charset.forName("utf-8"));
            // System.out.println(dirPath+cvsfile[j]);
            // 逐條讀取記錄，直至讀完
            while (r.readRecord()) {
                String regex = "(\\d+)";
                Pattern pattern = Pattern.compile(regex);
                Matcher matcher = pattern.matcher(cvsfile);
                String stockID = "";
                while (matcher.find()) {
                    stockID = matcher.group(1);
                }
                String date = sdf1.format(sdf.parse(ym + d + r.get(0)));
                File f = new File("E:\\shamrock\\project\\" + ym + "\\" + ym
                        + d + "\\" + cvsfile);
                if (!f.getParentFile().exists()) {
                    if (!f.getParentFile().mkdirs()) {
                        return;
                    }
                }
                String[] contents = { stockID, date, r.get(1), r.get(2),
                        r.get(3) };
                wr.writeRecord(contents);

            }

            r.close();
            wr.close();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

class LoadMysqlThread extends Thread {
    SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    public String dirPath;
    public String cvsfile;
    public String ym;
    public String d;
    public int dj;

    public LoadMysqlThread(String dirPath, String cvsfile, String ym, String d) {
        this.dirPath = dirPath;
        this.cvsfile = cvsfile;
        this.ym = ym;
        this.d = d;
        this.dj = dj;
    }
    //執行緒匯入資料庫
    @Override
    public void run() {
        try {
            
            Connection conn = null;
            String driver = "com.mysql.jdbc.Driver";
            String url = "jdbc:mysql://localhost:3306/tickdata";
            Class.forName(driver);
            conn = DriverManager.getConnection(url, "root", "123456");
            PreparedStatement pstmt = conn
                    .prepareStatement(" LOAD DATA INFILE 'E:/project/"
                            + ym
                            + "/"
                            + ym
                            + d
                            + "/"
                            + cvsfile
                            + "'   INTO TABLE d201505 FIELDS TERMINATED BY ',' ENCLOSED BY '\"' ESCAPED BY '\\\\' LINES TERMINATED BY '\\n' (stockID,date,price,buysell,volume);");
            pstmt.execute();
            conn.close();
        } catch (ClassNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

 

在此再說一個問題，比如上面的程式在自己電腦，匯入本地mysql資料庫沒問題，但問題來了，因為資料量較大，想讓別的電腦也來幫忙，然後存在我的電腦上，換句話說就是電腦A上的資料想匯入到電腦B上的mysql中，如果你用上面的命令，肯定會報錯找不到檔案，因為那語句指定的檔案是以mysql的安裝路徑為相對路徑的，所以如果不在B電腦的資料自然會報錯，於是此時你要LOCAL來指定是將本地A的資料匯入到B mysql中，語法如下：

LOAD DATA LOCAL INFILE 'E:/1.csv'   
INTO TABLE d201505 
FIELDS TERMINATED BY ',' 
ENCLOSED BY '\"' 
ESCAPED BY '\\' 
LINES TERMINATED BY '\n'
(stockID,date,price,buysell,volume);
 

也就是LOAD DATA INFILE中加入LOCAL.....

好了，如果有遇到我這樣問題的，希望能幫助到你哦！！！