MySQL 到 JSON 不一致提取

Question

我有一个包含 6 个表和大约 200 万行的 MySQL 数据库。

我想将所有数据迁移到 MongoDB。

我决定通过将 SQL 表转换为 JSON 并将其导入 MongoDB 来实现此目的。

我用 Golang 编写了一个程序来提取数据并将其输出为 JSON。

这是程序的主要功能:

func main() {
    // Open a database connection
    var err error
    db, err = sql.Open("mysql", "root:password@tcp(127.0.0.1:3306)/employees")
    checkErr(err)
    // Check if reachable
    if err = db.Ping(); err != nil {
        log.Fatal("Database is unreachable:", err)
    }
    // Populate variables with data
    err = populateVars()
    checkErr(err)
    // Marshal variables into JSON
    binaryJSON, err := json.Marshal(collection)
    checkErr(err)
    // Write JSON to a file
    err = writeStringToFile("/home/user01/Temporary/sql2data.json", string(binaryJSON))
    checkErr(err)
}

问题是输出不一致。

每次我运行该程序时，生成的文件都有不同的大小，并且缺少一些随机字段。

这可能是什么原因造成的？

这似乎不是程序逻辑的问题，因为一切都没有错误地执行，并且大多数字段都被填充得很好。

我是否阅读信息速度过快，以至于偶尔会丢失某些内容？

还是我还缺少其他东西？

编辑:

大部分工作发生在 populateVars() 函数调用中。

它有多个代码块，可执行给定的 SQL 查询并根据模式填充结构变量。

这是一个这样的 block :

rows, err = db.Query("SELECT emp_no, dept_emp.dept_no, dept_name, from_date, to_date FROM dept_emp JOIN departments ON departments.dept_no = dept_emp.dept_no;")
checkErr(err)
i := 0
for rows.Next() {
    var id int
    var depNumber string
    var depName string
    var fromDate string
    var toDate string
    var position = "Employee"
    err = rows.Scan(&id, &depNumber, &depName, &fromDate, &toDate,)
    // For debugging purposes:
    fmt.Println(id, depNumber, depName, fromDate, toDate, position, i)
    if err != nil {
        return err
    }
    for i := range collection {
        if collection[i].ID == id {
            collection[i].Departments = append(collection[i].Departments, Department{DepartmentNumber: depNumber, DepartmentName: depName, FromDate: fromDate, ToDate: toDate, Position: position})
            // For debugging purposes:
            fmt.Println(collection[i].Departments)
        }
    }
    i++
}

这是整个程序的 GitHub 链接: https://github.com/dchmie01/mysql_to_json/blob/master/main.go

编辑 2:

问题似乎与查询超时有关。

每个查询执行大约需要 10 分钟，但在大约 6 分钟后，我收到此错误，程序停止执行查询:

[mysql] 2017/04/29 17:35:16 packets.go:66: unexpected EOF
[mysql] 2017/04/29 17:35:16 packets.go:412: busy buffer
2017/04/29 17:35:16 driver: bad connection

在 MySQL 日志文件中它说:

2017-04-29T16:28:49.975805Z 102 [Note] Aborted connection 102 to db: 'employees' user: 'root' host: 'localhost' (Got timeout writing communication packets)

到目前为止，我尝试使用 MySQL 变量来禁用任何可能存在的超时，但没有成功。

我认为问题可能出在 Go 的 mysql 驱动程序上。

Answer 1

考虑使用Mysql SELECT INTO OUTFILE和 mongoiport --type csv相反。

该程序唯一要做的就是嵌入一对多和多对多文档，这可以通过聚合框架轻松完成。

一步一步的例子:

1. 从 mysql 导出 csv

   SELECT * from employees INTO OUTFILE '/tmp/employees.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   SELECT * from salaries INTO OUTFILE '/tmp/salaries.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   SELECT * from titles INTO OUTFILE '/tmp/titles.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   SELECT * from departments INTO OUTFILE '/tmp.departments.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   SELECT * from dept_emp INTO OUTFILE '/tmp/dept_emp.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   SELECT * from dept_manager INTO OUTFILE '/tmp/dept_manager.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"';
   

2. 将 csv 导入 mongo(根据您的架构定义“字段规范”，请参阅员工字段规范示例)

   mongoimport -d <dbname> -c tmp_employees -f 'id.int32(),birth.date(2006-01-02),first_name.string(),last_name.string(),gender.string(),hire_date.date(2006-01-02)' --columnsHaveTypes --type csv --file /tmp/employees.csv --drop 
   mongoimport -d <dbname> -c tmp_salaries -f 'field spec' --columnsHaveTypes --type csv --file /tmp/salaries.csv --drop 
   mongoimport -d <dbname> -c tmp_titles -f 'field spec' --columnsHaveTypes --type csv --file /tmp/titles.csv --drop 
   mongoimport -d <dbname> -c tmp_departments -f 'field spec' --columnsHaveTypes --type csv --file /tmp/departments.csv --drop 
   mongoimport -d <dbname> -c tmp_dept_emp -f 'field spec' --columnsHaveTypes --type csv --file /tmp/dept_emp.csv --drop 
   mongoimport -d <dbname> -c tmp_dept_manager -f 'field spec' --columnsHaveTypes --type csv --file /tmp/dept_manager.csv --drop 
   

3. 从 mongo shell 嵌入数据

   db.tmp_employees.aggregate([
       // 1-to-many joins
       {$lookup: {
           from: 'tmp_salaries',
           localField: 'id',
           foreignField: 'emp_no',
           as: 'salaries'
       }},
       {$lookup: {
           from: 'tmp_titles',
           localField: 'id',
           foreignField: 'emp_no',
           as: 'titles'
       }},
       // many-to-many joins
       {$lookup: {
           from: 'tmp_dept_emp',
           localField: 'id',
           foreignField: 'emp_no',
           as: 'dept_emp'
       }},
       {$lookup: {
           from: 'tmp_dept_manager',
           localField: 'id',
           foreignField: 'emp_no',
           as: 'dept_manager'
       }},
       {$unwind: { path: '$dept_emp', preserveNullAndEmptyArrays: true }},
       {$lookup: {
           from: 'tmp_departments',
           localField: 'dept_emp.dept_no',
           foreignField: 'dept_no',
           as: 'dept_emp_deps'
       }},    
       {$unwind: { path: '$dept_emp_deps', preserveNullAndEmptyArrays: true }},
       {$group: {
           _id: '$_id',
           root: {$first: '$$ROOT'},
           dept_manager: {$first: '$dept_manager'},
           departments_emp: {$push: {
               department_number: '$dept_emp.emp_no',
               department_name: '$dept_emp_deps.dept_name',
               from_date: '$dept_emp.from_date',
               to_date: '$dept_emp.to_date',
               position: '$dept_emp.position'
           }},
       }},
       {$unwind: { path: '$dept_manager', preserveNullAndEmptyArrays: true }},
       {$lookup: {
           from: 'tmp_departments',
           localField: 'dept_manager.dept_no',
           foreignField: 'dept_no',
           as: 'dept_manager_deps'
       }},    
       {$unwind: { path: '$dept_manager_deps', preserveNullAndEmptyArrays: true }},
       {$group: {
           _id: '$_id',
           root: {$first: '$root'},
           departments_emp: {$first: '$departments_emp'},
           departments_manager: {$push: {
               department_number: '$dept_manager.emp_no',
               department_name: '$dept_manager_deps.dept_name',
               from_date: '$dept_manager.from_date',
               to_date: '$dept_manager.to_date',
               position: '$dept_manager.position'
           }},
       }},
       // combine departments to a single array
       {$project: {
           root: 1,
           departments_all: {$concatArrays: [ "$departments_emp", "$departments_manager" ] }
       }},
       //final reshape
       {$project: {
           id: '$root.id',
           birth_date: '$root.birth_date',
           first_name: '$root.first_name',
           last_name: '$root.last_name',
           gender: '$root.gender',
           hire_date: '$root.hire_date',
           salaries: '$root.salaries',
           titles: '$root.titles',
           departments: {$filter: {
               input: "$departments_all",
               as: "departments",
               cond: { $ne: [ "$$departments", {} ] }}}
       }},
       { $out : "employees" }
   ])
   

4. 从 mongo shell 中删除导入的集合

   db.tmp_employees.drop();
   db.tmp_salaries.drop();
   db.tmp_titles.drop();
   db.tmp_departments.drop();
   db.tmp_dept_emp.drop();
   db.tmp_dept_manager.drop();