写在开头

最近在做elasticsearch相关的开发，有这么一个需求，需要将mysql指定表的数据导入到elasticsearch的索引中，这个需求似乎比较容易，大致思路就是从遍历表中的每一行数据，转换成json，然后通过es的api接口提交到es中。

这个似乎是一个比较简单需求，只是有一点这是一个通用的操作，程序不需要关心遥操作的是什么样的表，包含哪些字段等等。

第一版

stackoverflow上已经有了一个比较好的解决方案：
这里只使用了 go-sql-driver/mysql

func getJSON(sqlString string) (string, error) {
  rows, err := db.Query(sqlString)
  if err != nil {
      return "", err
  }
  defer rows.Close()
  columns, err := rows.Columns()
  if err != nil {
      return "", err
  }
  count := len(columns)
  tableData := make([]map[string]interface{}, 0)
  values := make([]interface{}, count)
  valuePtrs := make([]interface{}, count)
  for rows.Next() {
      for i := 0; i < count; i++ {
          valuePtrs[i] = &values[i]
      }
      rows.Scan(valuePtrs...)
      entry := make(map[string]interface{})
      for i, col := range columns {
          var v interface{}
          val := values[i]
          b, ok := val.([]byte)
          if ok {
              v = string(b)
          } else {
              v = val
          }
          entry[col] = v
      }
      tableData = append(tableData, entry)
  }
  jsonData, err := json.Marshal(tableData)
  if err != nil {
      return "", err
  }
  fmt.Println(string(jsonData))
  return string(jsonData), nil 
}

代码出自 Dumping MySQL tables to JSON with Golang

第二版

这段代码基本是可以满足需求了，但似乎有一个问题，那就是字段对应的字段的类型并没有保留，全都变成了字符串类型，这里找到了一个比较可行的解决方案：

在这个mysql driver的github里有这么个issue，Returned values are always []byte

That’s a specialty of MySQL: you have to use prepared statements to get the native types. MySQL has two protocols, one transmits everything as text, the other as the “real” type. And that binary protocol is only used when you use prepared statements. The driver is pretty much powerless to enforce a protocol and the text protocol takes less resources on the server.

This may help you:

stmt, err := db.Prepare(sqlString)
if err != nil { …… }
defer stmt.Close()
rows, err := stmt.Query()

这个描述比较清楚了，按照这个逻辑，我把代码修改成了如下形式:

    query := fmt.Sprintf("%s LIMIT %d OFFSET %d", handler.querySQL, limit, offset)

    stmt, err := db.Prepare(query)
    if err != nil {
        panic(err.Error())
    }
    defer stmt.Close()

    rows, err := stmt.Query()
    if err != nil {
        panic(err.Error())
    }
    defer rows.Close()

    columns, err := rows.Columns()
    if err != nil {
        panic(err.Error())
    }

    count := len(columns)
    values := make([]interface{}, count)
    scanArgs := make([]interface{}, count)

    for i := range values {
        scanArgs[i] = &values[i]
    }

    for rows.Next() {
        err := rows.Scan(scanArgs...)
        if err != nil {
            log.Fatalf("Sacn data error: %s", err.Error())
            continue
        }

        entry := make(map[string]interface{})
        for i, col := range columns {
            v := values[i]

            b, ok := v.([]byte)
            if ok {
                entry[col] = string(b)
            } else {
                entry[col] = v
            }
        }

        // 序列化数据
        b, err := json.Marshal(entry)
        if err != nil {
            log.Fatal(err.Error())
            continue
        }

        handler.docChan <- &document{
            // 这里取表中的第一个字段计算文档id
            ID:   handler.FormatDocumentID(handler.dbName, handler.table, entry[columns[0]]),
            Data: b,
        }
    }

测试来看，似乎只是解决了部分类型问题，整形的数据类型可以被保留下来，但是似乎浮点型字段还是字符串，还在继续探索中。

未完待续…….