golang框架在数据处理中的应用：大数据和大文件处理

import (
    "fmt"

    "github.com/apache/spark-go/spark"
)

func main() {
    // 创建 Spark 上下文
    sc, err := spark.NewContext("local[*]", "Spark Example")
    if err != nil {
        fmt.Println("Error creating Spark context:", err)
        return
    }

    // 读取 CSV 文件
    rdd := sc.TextFile("stocks.csv")

    // 使用 map 转换将每一行的字段拆分到数组中
    stockData := rdd.Map(func(line string) []string {
        return strings.Split(line, ",")
    })

    // 转换为元组，其中第一列为股票名称，第二列为股票价格
    stockInfo := stockData.Map(func(fields []string) (string, float64) {
        return fields[0], strconv.ParseFloat(fields[1], 64)
    })

    // 根据股票名称对价格进行分组和求和
    groupedData := stockInfo.GroupByKey().MapGroups(func(key string, values chan float64) float64 {
        var total float64
        for price := range values {
            total += price
        }
        return total
    })

    // 将结果保存在文件
    groupedData.SaveAsTextFile("output.txt")
}

import (
    "context"
    "fmt"
    "io"

    "cloud.google.com/go/hadoopmapreduce/apiv1/hadoopmapreducepb"
    "cloud.google.com/go/hadoopmapreduce/apiv1/hadoopmapreducepb/hadoopmapreducepbtesting"
)

func main() {
    ctx := context.Background()
    mr := hadoopmapreducepbtesting.NewMockJobServiceClient()

    request := &hadoopmapreducepb.JobSubmissionRequest{
        JobName: "Web Log Analysis",
        Driver: &hadoopmapreducepb.JobSubmissionRequest_MainClass{
            MainClass: "org.apache.hadoop.mapred.JobControl",
        },
        Args: []string{
            "yarnClasspath",
            "hadoop",
            "./web_log_mapper.py",
            "./web_log_reducer.py",
            "/input/web_logs.txt",
            "/output/page_counts.txt",
        },
        Configuration: &hadoopmapreducepb.Configuration{
            Properties: map[string]string{
                "mapred.mapper.class": "WebLogMapper",
                "mapred.reducer.class": "WebLogReducer",
                "mapred.output.dir": "/output/page_counts.txt",
            },
        },
    }

    // 执行作业
    job, err := mr.SubmitJob(ctx, request)
    if err != nil {
        fmt.Println("Error submitting job:", err)
        return
    }

    // 跟踪作业状态
    for {
        response, err := mr.GetJob(ctx, &hadoopmapreducepb.JobGetRequest{JobId: job.GetJobId()})
        if err != nil {
            fmt.Println("Error getting job status:", err)
            return
        }
        status := response.GetJobStatus().GetState().String()
        if status == "KILLED" || status == "FAILED" {
            fmt.Println("Job failed. Status:", status)
            return
        } else if status == "SUCCEEDED" {
            fmt.Println("Job succeeded")
            break
        }
    }
}

import (
    "context"
    "fmt"

    "cloud.google.com/go/bigquery"
)

func main() {
    ctx := context.Background()

    // 客户端创建
    client, err := bigquery.NewClient(ctx, "your-project-id")
    if err != nil {
        fmt.Println("Error creating client:", err)
        return
    }

    query := client.Query(
        `SELECT
            device_id,
            location,
            timestamp,
            value
        FROM 
            ` + "`bigquery-public-data.iot_demo.sensordata`" + `
        WHERE 
            value > (
                SELECT 
                    AVG(value) + STDDEV(value)
                FROM 
                    ` + "`bigquery-public-data.iot_demo.sensordata`" + `
            )
        ORDER BY 
            value DESC
        LIMIT 
            10`,
    )
    
    rows, err := query.Read(ctx)
    if err != nil {
        fmt.Println("Error querying data:", err)
        return
    }

    fmt.Println("Anomalies:")
    for rows.Next() {
        var deviceID string
        var location string
        var timestamp bigquery.NullTimestamp
        var value float64

        if err := rows.Scan(&deviceID, &location, &timestamp, &value); err != nil {
            fmt.Println("Error scanning row:", err)
            continue
        }

        fmt.Printf("%s %s %s %f\n", deviceID, location, timestamp.String, value)
    }
}