golang 将sql.Rows转换为地图

Posted 2021-05-24

// Convert an sql.Rows to a map

func SqlToMap(rows *sql.Rows) {

        columns, err := rows.Columns()

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

        results := make([]map[string]interface{}, 0)

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

        for rows.Next() {
                err = rows.Scan(scanArgs...)
                if (err != nil) {
                        panic(err)
                }

                record := make(map[string]interface{})

                for i, col := range values {
                        if col != nil {
                                switch col.(type) {
                                case bool:
                                        record[columns[i]] = col.(bool)
                                case int:
                                        record[columns[i]] = col.(int)
                                case int64:
                                        record[columns[i]] = col.(int64)
                                case float64:
                                        record[columns[i]] = col.(float64)
                                case string:
                                        record[columns[i]] = col.(string)
                                case time.Time:
                                        record[columns[i]] = col.(time.Time)
                                case []byte:
                                        record[columns[i]] = string(col.([]byte))
                                default:
                                        record[columns[i]] = col
                                }
                        }
                }

                results = append(results, record)
        }

        s, _ := json.Marshal(results)
        fmt.Println(string(s))
}

database/sql rows.scan 在 350K 行后挂起

【中文标题】database/sql rows.scan 在 350K 行后挂起

【英文标题】：database/sql rows.scan hangs after 350K rows

【发布时间】：2020-08-20 21:55:37

【问题描述】：

我的任务是从 Oracle 数据库中提取数据，我正在尝试提取大量数据 > 6MM 记录，其中包含 100 列进行处理。

需要将数据转换为地图。我成功地在不到 35 秒的时间内处理了 35 万条记录。之后，服务器挂起并且不再继续。 有没有办法可以根据行大小对它们进行批处理或批处理以释放我的空间。

func FetchUsingGenericResult(ctx context.Context, dsConnection *string, sqlStatement string) (*entity.GenericResultCollector, error) 
    columnTypes := make(map[string]string)
    var resultCollection entity.GenericResultCollector
    db, err := sql.Open("godror", *dsConnection)
    if err != nil 
        return &resultCollection, errors.Wrap(err, "error connecting to Oracle")
    
    log := logger.FromContext(ctx).Sugar()
    log.Infof("start querying from Oracle at :%v", time.Now())
    rows, err := db.Query(sqlStatement, godror.FetchRowCount(defaultFetchCount))
    if err != nil 
        return &resultCollection, errors.Wrap(err, "error querying")
    
    objects, err := rows2Strings(ctx, rows)
    log.Infof("total Rows converted are :%v by %v", len(*objects), time.Now())
    resultCollection = entity.GenericResultCollector
        Columns: columnTypes,
        Rows:    objects,
    
    return &resultCollection, nil


func rows2Strings(ctx context.Context, rows *sql.Rows) (*[]map[string]string, error) 
    result := make(map[string]string)
    resultsSlice := []map[string]string
    fields, err := rows.Columns()
    if err != nil 
        return nil, err
    
    log := logger.FromContext(ctx).Sugar()
    waitGroup, ctx := errgroup.WithContext(ctx)
    counter := 0
    for rows.Next() 
        counter++
        if counter%defaultFetchCount == 0 
            log.Infof("finished converting %v rows by %v", counter, time.Now())
        
        waitGroup.Go(func() error 
            result, err = row2mapStr(rows, fields)
            if err != nil 
                return err
            
            resultsSlice = append(resultsSlice, result)
            return nil
        )
        if err := waitGroup.Wait(); err != nil 
            return nil, err
        
    
    return &resultsSlice, nil


func row2mapStr(rows *sql.Rows, fields []string) (resultsMap map[string]string, err error) 
    result := make(map[string]string)
    scanResultContainers := make([]interface, len(fields))
    for i := 0; i < len(fields); i++ 
        var scanResultContainer interface
        scanResultContainers[i] = &scanResultContainer
    
    if err := rows.Scan(scanResultContainers...); err != nil 
        return nil, err
    
    for ii, key := range fields 
        rawValue := reflect.Indirect(reflect.ValueOf(scanResultContainers[ii]))
        // if row is null then as empty string
        if rawValue.Interface() == nil 
            result[key] = ""
            continue
        

        if data, err := value2String(&rawValue); err == nil 
            result[key] = data
         else 
            return nil, err
        
    
    return result, nil


func value2String(rawValue *reflect.Value) (str string, err error) 
    aa := reflect.TypeOf((*rawValue).Interface())
    vv := reflect.ValueOf((*rawValue).Interface())
    switch aa.Kind() 
    case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
        str = strconv.FormatInt(vv.Int(), 10)
    case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
        str = strconv.FormatUint(vv.Uint(), 10)
    case reflect.Float32, reflect.Float64:
        str = strconv.FormatFloat(vv.Float(), 'f', -1, 64)
    case reflect.String:
        str = vv.String()
    case reflect.Array, reflect.Slice:
        switch aa.Elem().Kind() 
        case reflect.Uint8:
            data := rawValue.Interface().([]byte)
            str = string(data)
            if str == "\x00" 
                str = "0"
            
        default:
            err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
        
    // time type
    case reflect.Struct:
        if aa.ConvertibleTo(timeType) 
            str = vv.Convert(timeType).Interface().(time.Time).Format(time.RFC3339Nano)
         else 
            err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
        
    case reflect.Bool:
        str = strconv.FormatBool(vv.Bool())
    case reflect.Complex128, reflect.Complex64:
        str = fmt.Sprintf("%v", vv.Complex())
    default:
        err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
    
    return

有没有人遇到过类似的问题？

修改代码如下：

func FetchUsingGenericResult(ctx context.Context, dsConnection *string, sqlStatement string) (*entity.GenericResultCollector, error) 
    columnTypes := make(map[string]string)
    var resultCollection entity.GenericResultCollector
    db, err := sql.Open("godror", *dsConnection)
    if err != nil 
        return &resultCollection, errors.Wrap(err, "error connecting to Oracle")
    
    log := logger.FromContext(ctx).Sugar()
    log.Infof("start querying from Oracle at :%v", time.Now())
    rows, err := db.Query(sqlStatement, godror.FetchRowCount(defaultFetchCount))
    if err != nil 
        return &resultCollection, errors.Wrap(err, "error querying")
    
    objects, err := rows2Strings(ctx, rows)
    log.Infof("total Rows converted are :%v by %v", len(*objects), time.Now())
    resultCollection = entity.GenericResultCollector
        Columns: columnTypes,
        Rows:    objects,
    
    return &resultCollection, nil


func rows2Strings(ctx context.Context, rows *sql.Rows) (*[]map[string]string, error) 
    result := make(map[string]string)
    resultsSlice := []map[string]string
    fields, err := rows.Columns()
    if err != nil 
        return nil, err
    
    log := logger.FromContext(ctx).Sugar()
    counter := 0
    for rows.Next() 
        counter++
        if counter%defaultFetchCount == 0 
            log.Infof("finished converting %v rows by %v", counter, time.Now())
        
        result, err = row2mapStr(rows, fields)
        if err != nil 
            return nil, err
        
        resultsSlice = append(resultsSlice, result)
    
    return &resultsSlice, nil


func row2mapStr(rows *sql.Rows, fields []string) (resultsMap map[string]string, err error) 
    result := make(map[string]string)
    scanResultContainers := make([]interface, len(fields))
    for i := 0; i < len(fields); i++ 
        var scanResultContainer interface
        scanResultContainers[i] = &scanResultContainer
    
    if err := rows.Scan(scanResultContainers...); err != nil 
        return nil, err
    
    for ii, key := range fields 
        rawValue := reflect.Indirect(reflect.ValueOf(scanResultContainers[ii]))
        // if row is null then as empty string
        if rawValue.Interface() == nil 
            result[key] = ""
            continue
        

        if data, err := value2String(&rawValue); err == nil 
            result[key] = data
         else 
            return nil, err
        
    
    return result, nil


func value2String(rawValue *reflect.Value) (str string, err error) 
    aa := reflect.TypeOf((*rawValue).Interface())
    vv := reflect.ValueOf((*rawValue).Interface())
    switch aa.Kind() 
    case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
        str = strconv.FormatInt(vv.Int(), 10)
    case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
        str = strconv.FormatUint(vv.Uint(), 10)
    case reflect.Float32, reflect.Float64:
        str = strconv.FormatFloat(vv.Float(), 'f', -1, 64)
    case reflect.String:
        str = vv.String()
    case reflect.Array, reflect.Slice:
        switch aa.Elem().Kind() 
        case reflect.Uint8:
            data := rawValue.Interface().([]byte)
            str = string(data)
            if str == "\x00" 
                str = "0"
            
        default:
            err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
        
    // time type
    case reflect.Struct:
        if aa.ConvertibleTo(timeType) 
            str = vv.Convert(timeType).Interface().(time.Time).Format(time.RFC3339Nano)
         else 
            err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
        
    case reflect.Bool:
        str = strconv.FormatBool(vv.Bool())
    case reflect.Complex128, reflect.Complex64:
        str = fmt.Sprintf("%v", vv.Complex())
    default:
        err = fmt.Errorf("Unsupported struct type %v", vv.Type().Name())
    
    return

【问题讨论】：

waitGroup 是什么东西？这是顺序行处理代码，有什么需要？也许这就是问题所在？

@BurakSerdar waitGroup 被编码，它强制执行顺序处理。他在 Go 之后立即等待。这意味着它只会等待一个 goroutine。对我来说，更大的问题是代码中内置的泛化量。识别列类型并定位这些类型的小查询将显着减少代码大小。

@praveent 我在代码中看不到任何可能导致所描述行为的内容，除了 waitGroup 的东西，也许还有一些服务器端问题。我的第一直觉是删除与等待组相关的代码。在 for 循环中从数据库中读取行不需要单独的 goroutine。

@BurakSerdar 我同意。 waitGroup 不会向代码添加任何内容。它的编写方式是一种迂回的方式来编写顺序代码。

这是服务器端的空间问题。我在扫描时更改了处理数据的逻辑。谢谢@MaximKosov

【参考方案1】：

由于服务器内存有限，阵列可能会卡住而无法继续进行。 我已经开始处理扫描的数据，这解决了我的问题。