Gunakan Hadoop dalam bahasa Go untuk mencapai pemprosesan data besar yang cekap

Dengan peningkatan jumlah data, pemprosesan data besar telah menjadi salah satu topik yang paling membimbangkan dalam industri hari ini. Sebagai rangka kerja pengkomputeran teragih sumber terbuka, Hadoop telah menjadi standard de facto untuk pemprosesan data besar. Dalam artikel ini, kami akan memperkenalkan cara menggunakan Hadoop dalam bahasa Go untuk mencapai pemprosesan data besar yang cekap.

Mengapa menggunakan Hadoop dalam bahasa Go?

Pertama sekali, bahasa Go ialah bahasa pengaturcaraan baharu yang dibangunkan oleh Google Ia mempunyai keupayaan pengurusan memori dan pengaturcaraan serentak yang cekap, mudah untuk ditulis dan mempunyai kelajuan kompilasi yang pantas. Ia sangat sesuai untuk membangunkan pelayan yang cekap program. Kedua, Hadoop menyediakan keupayaan pemprosesan data teragih yang berkuasa dan boleh memproses sejumlah besar data dengan cekap Ia adalah rangka kerja perisian percuma sumber terbuka yang boleh membina sistem pengkomputeran berskala besar dengan cepat.

Bagaimana untuk menggunakan Hadoop dalam bahasa Go?

Bahasa Go sendiri tidak menyokong pengaturcaraan Hadoop asli, tetapi kami boleh menggunakan ciri Cgo bahasa Go untuk memanggil antara muka C/C++ yang disediakan oleh Hadoop untuk melengkapkan akses dan pengendalian Hadoop. Cgo ialah ciri yang disediakan oleh bahasa Go yang membenarkan pengaturcara memanggil program C/C++ dalam bahasa Go untuk menyelesaikan tugasan tertentu.

Pertama, kita perlu memasang Hadoop dan perpustakaan pembangunan C/C++ yang sepadan secara tempatan. Untuk pengedaran Linux biasa, anda boleh memasang terus perpustakaan bergantung yang berkaitan melalui pengurus pakej, seperti libhadoop2.10.1, hadoop-c++-libs, dsb. Jika anda berada di bawah sistem Windows, anda boleh menyusun perpustakaan C/C++ yang sepadan melalui rantai alat penyusunan di bawah Windows.

Seterusnya, gunakan ciri Cgo dalam program bahasa Go untuk memulakan tugas pengkomputeran yang diedarkan Hadoop. Pelaksanaan khusus adalah seperti berikut:

package main

// #include "hdfs.h"
import "C"

import (
    "fmt"
    "unsafe"
)

func main() {
    const hadoopConfDir = "/etc/hadoop/conf"
    const hadoopAddress = "hdfs://localhost:9000"
    var buf [64]C.char

    C.hdfsGetDefaultConfigPath(&buf[0], 64)
    confDir := C.GoString(&buf[0])
    if confDir == "" {
        confDir = hadoopConfDir
    }

    fs := C.hdfsNew(hadoopAddress, "default")
    defer C.hdfsDisconnect(fs)

    if fs == nil {
        panic(fmt.Errorf("Could not connect to Hadoop Namenode at: %s", hadoopAddress))
    }

    basePath := C.CString("/")
    defer C.free(unsafe.Pointer(basePath))

    fileInfo, _ := C.hdfsListDirectory(fs, basePath, nil)

    for i := 0; fileInfo[i] != nil; i++ {
        fileInfoEntry := fileInfo[i]
        fmt.Println(C.GoString(fileInfoEntry.mName))
    }

    C.hdfsFreeFileInfo(fileInfo, 1)
}

Kod di atas menunjukkan cara memulakan tugas pengkomputeran teragih Hadoop dalam program bahasa Go. Antaranya, kita perlu cuba menggunakan fungsi C hdfsGetDefaultConfigPath yang disediakan dalam perpustakaan libhdfs dalam program untuk mendapatkan laluan lalai fail konfigurasi Hadoop. Jika pemerolehan gagal, laluan yang ditentukan oleh pemalar hadoopConfDir digunakan sebagai laluan ke fail konfigurasi.

Seterusnya, kami menggunakan fungsi hdfsNew untuk mencipta objek sistem fail Hadoop fs Jika penciptaan gagal, ini bermakna pelayan Hadoop tidak boleh disambungkan dan program akan ralat serta-merta. Seterusnya, kami melaksanakan fungsi hdfsListDirectory untuk menyenaraikan semua fail dan direktori dalam direktori akar sistem fail Hadoop dan mengeluarkannya dalam konsol.

Akhir sekali, kita perlu melepaskan memori secara manual dan memanggil fungsi hdfsDisconnect untuk menutup objek sistem fail hdfs. Ambil perhatian bahawa untuk memperuntukkan dan melepaskan memori Cgo dengan betul, apabila menggunakan penunjuk objek bahasa C, anda perlu menggunakan C.CString atau C.GoString dan fungsi khusus Cgo lain untuk menukar rentetan bahasa Go kepada rentetan bahasa C semasa menggunakan C. fungsi bebas melepaskan ruang memori C yang diminta.

Menggunakan Hadoop untuk pengisihan data besar

Dalam pemprosesan data berskala besar sebenar, selalunya perlu mengisih data untuk mengoptimumkan prestasi pemprosesan program. Demonstrasi berikut menggunakan Hadoop dalam bahasa Go untuk pengisihan data besar:

package main

// #include "hdfs.h"
import "C"

import (
    "fmt"
    "unsafe"
)

func main() {
    const hadoopAddress = "hdfs://localhost:9000"
    var buf [64]C.char

    C.hdfsGetDefaultConfigPath(&buf[0], 64)
    confDir := C.GoString(&buf[0])
    if confDir == "" {
        panic(fmt.Errorf("Could not find Hadoop configuration"))
    }

    fs := C.hdfsNew(hadoopAddress, "default")
    defer C.hdfsDisconnect(fs)

    const inputPath = "/input"
    const outputPath = "/output"

    inputPathC := C.CString(inputPath)
    outputPathC := C.CString(outputPath)
    defer C.free(unsafe.Pointer(inputPathC))
    defer C.free(unsafe.Pointer(outputPathC))

    sortJobConf := C.hdfsNewJobConf()
    defer C.hdfsDeleteJobConf(sortJobConf)

    C.hdfsConfSet(sortJobConf, C.CString("mapred.reduce.tasks"), C.CString("1"))

    const mapperFunc = `package main
      import (
          "bufio"
          "fmt"
          "os"
          "sort"
          "strings"
      )

      func main() {
          scanner := bufio.NewScanner(os.Stdin)
          var lines []string

          for scanner.Scan() {
              lines = append(lines, scanner.Text())
          }

          sort.Strings(lines)

          for _, str := range lines {
              fmt.Println(str)
          }
      }
    `

    const reducerFunc = ""

    C.hdfsRunStreaming(fs, sortJobConf, 1,
        &inputPathC, 1,
        &outputPathC, 1,
        (*C.char)(unsafe.Pointer(&[]byte(mapperFunc)[0])), C.uint(len(mapperFunc)),
        (*C.char)(unsafe.Pointer(&[]byte(reducerFunc)[0])), C.uint(len(reducerFunc)),
    )

    fmt.Println("Finished sorting")
}

Kod di atas menunjukkan kaedah menggunakan Hadoop dalam bahasa Go untuk pengisihan data besar. Mula-mula, kami mencipta objek conf kerja Hadoop sortJobConf dan menetapkan parameter mapred.reduce.tasks mengikut keperluan Di sini ia ditetapkan kepada 1, yang bermaksud hanya satu tugasan pengurangan dilaksanakan.

Seterusnya, kami mentakrifkan fungsi mapperFunc untuk membaca fail input dan mengisihnya mengikut saiz rentetan. reducerFunc ialah fungsi kosong, menunjukkan bahawa tiada langkah pengurangan dalam tugas ini.

Akhir sekali, kami menggunakan fungsi hdfsRunStreaming untuk memulakan pengiraan aliran Hadoop, memasukkan sortJobConf sebagai parameter dan menentukan laluan fail input dan output serta fungsi pemeta dan pengurang untuk menyelesaikan tugas data menyusun.

Ringkasan

Artikel ini memperkenalkan secara ringkas cara menggunakan Hadoop dalam bahasa Go untuk pemprosesan data besar. Mula-mula, kami memperkenalkan kaedah menggunakan ciri Cgo untuk memanggil antara muka C/C++ Hadoop dalam bahasa Go. Seterusnya, kami menunjukkan cara menggunakan Hadoop untuk pengisihan data besar. Melalui pengenalan artikel ini, pembaca boleh mempelajari cara menggunakan bahasa Go dan Hadoop untuk pemprosesan data besar yang cekap.

Atas ialah kandungan terperinci Gunakan Hadoop dalam bahasa Go untuk mencapai pemprosesan data besar yang cekap. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!