数据工程师关于Go与Python的比较（第1部分）

介绍

探索golang —我们可以放弃Python吗？ 而且我们是否（如"经常处理大量数据的人们"那样）最终找到了go的用例？ 第1部分探索了Python和go之间的高级差异，并给出了这两种语言的具体示例，旨在回答基于Apache Beam和Google Dataflow的实际示例。

我曾经在此博客上使用过两次Apache Beam：这是一个用于批处理和流式用例的统一编程模型，可以处理令人愉悦的并行工作负载，并允许许多自定义I / O和其他连接器，以及在多种执行平台上运行，最著名的是Flink，Spark和Google Cloud的Dataflow。

您可以将其用于流式处理或批处理数据以对其进行分析，运行ETL并对其进行充实-您可以对其进行命名。 实际上，几年前我在这里使用过Beam。

现在，如果您要访问Beam的网站，则会发现以下精美图片：

它告诉您Beam是用Java或Python编写的，其他所有内容都属于"其他语言"。

但是，有一个"实验性"（GitHub）保证我们使用go代替Python-这就是让我兴奋的地方。

下一节将很少尝试给出一个"简短的"（约3,000个单词）的介绍，并根据不同的处理数据工程的特定用例以及使用多线程应用程序时将其与Python进行比较。 请注意，这绝不是完整的比较。

这是第2部分（共1部分），重点介绍两种语言以及Apache Beam SDK的当前状态。 在第2部分（即将推出）中，我们将探索如何使用Beam Go SDK，在GCP上将面临哪些局限性以及数据工程师的发展方向。

所有示例代码都可以在GitHub上找到。

Go vs.Python

如果您不熟悉，请允许我引用维基百科：

Go是由Robert Griesemer，Rob Pike和Ken Thompson在Google设计的一种静态类型的编译语言。 Go在语法上类似于C，但是具有内存安全性，垃圾回收，结构化类型和CSP样式的并发性。

https://en.wikipedia.org/wiki/Go_(programming_language)

Go有很多概念，应该让您为处理数据的并行执行框架而兴奋。 在以下各节中，我们将探索一些精心挑选的概念和示例，并以更长，更深入的并发示例作为结束。

速度

与解释型语言相反，go是一种编译器，它比Python更快。 它也倾向于真正快速地编译事物，并且比起尝试用Python进行同样的操作，将编译后的代码交付给最终用户要容易得多。

您可以在下面找到一些动手示例。

静态类型

相对于Python的动态类型而言，静态类型避免了在正确的时机使用正确的数据类型的麻烦。

我并不是试图在静态和动态类型与Python的复杂性之间进行一般的比较，而只是根据我的经验，特别是在数据空间中，每天进行比较。

在数据驱动的项目中，您将不断发现自己要处理非常严格的数据类型定义，通常是不可避免地将所有公司数据存储在大型RDMB（例如Oracle）中强制执行严格键入的区域的工件（当然是这样）。

诸如pySpark甚至pandas之类的基于Python的框架附带了自己的抽象层，用于提供类型（请参见dtypes）。

以使用为NYS Yellow Cab数据<0>为pySpark（一种流行的数据处理框架，为Python（以及其他）提供）编写的代码为例。

首先，将CSV文件读入RDD（一个"弹性的，分布式数据集"），并在数据集中的每一行上应用一些转换lambda。

# Read datardd = sc.textFile('data/yellow_tripdata_2019-01.csv')# Parse the RDDrdd = rdd.map(lambda r: r.split(','))\ .map(lambda r: (r<10>,r<13>)) # Take 'fare_amount' and 'tip_amount'rdd = rdd.filter(lambda r: 'fare_amount' not in r) # filter headerrdd.take(1)

产生：

<('7', '1.65'), ('14', '1'), ('4.5', '0'), ('3.5', '0'), ('52', '0')>

如果我们查看其中一个元组：

type(rdd.take(1)<0><0>)

我们将str作为数据类型。

现在，如果我们要减少此费用以总结出租车费和小费金额：

def sum_fares(fare, tip):   return fare + tiprdd.map(lambda r: sum_fares(*r)).take(5)

如以上输出中的引号所示，结果是一个串联字符串的列表。

<'71.65', '141', '4.50', '3.50', '520'>

而不是数学上正确的：

rdd.map(lambda r: sum_fares(*)).take(5)# <8.65, 15.0, 4.5, 3.5, 52.0>

较新版本的Python支持类型提示，而Python解释器则跟踪变量类型。 但是，如以上示例所示，我个人很难维护一致，可读和可维护的代码库，尤其是在复杂的应用程序上。

另一方面，go是静态类型的。

package mainimport (   "fmt"   "reflect")func main() {   // Implicit   str := "A String"   fmt.Printf("%s is %s\n", str, reflect.TypeOf(str))   // Explicit   var x float64 = 3.14   fmt.Printf("%v is %s\n", x, reflect.TypeOf(x))}

将产生：

go run test.go

A String is string3.14 is float64

鉴于此：

str = x

无法编译：

go run testvar.go

# command-line-arguments./test.go:15:6: cannot use x (type float64) as type string in assignment

Go不支持泛型，但Python也不支持。

我跳过了go的空接口{}概念，以支持任意值并处理未知类型； 可能在需要弱类型抽象的几种极端情况下使用此概念

<0> Spark的SQL接口将推断字符串类型，并允许对字符串进行数学运算（如果它们一致）

接口和结构

Python确实具有类结构（并且我已经广泛使用了它），而go使用了结构和接口（这是一个可怕的简化）。 go没有继承，并且依赖于接口和组成。

在数据世界中，拥有严格的类结构（例如抽象转换，统计模型或仅是简单的旧数据结构）既是痛苦，也是诅咒。

Python广泛使用可以容纳任意键值对和嵌套结构且与JSON语法相似的字典，以及用于数据结构化抽象的已定义类。 几乎每个数据框架都附带有自己的架构类。 从理论上讲，go可以避免通过接口（用于标准），静态类型化（以确保在正确的时机使用正确的数据类型）以及用于定义结构和逻辑的结构的组合来避免这种情况。

这是一个有用的示例，该示例使用接口CoordinateData和函数calculateDistance来计算世界上最差的GIS分析平台中两个坐标元组之间的距离：）

package mainimport (	"fmt"	"math")// A Resource we're trying to accesstype CoordinateData interface {	calculateDistance(latTo, lonTo float64) float64}

然后，我们实现地理空间数据和Haversine函数（以近似地球上的距离）：

type GeospatialData struct {	lat, lon float64}const earthRadius = float64(6371)func (d GeospatialData) calculateDistance(latTo, lonTo float64) float64 {	// Haversine distance	var deltaLat = (latTo - d.lat) * (math.Pi / 180)	var deltaLon = (lonTo - d.lon) * (math.Pi / 180)		var a = math.Sin(deltaLat / 2) * math.Sin(deltaLat / 2) + 		math.Cos(d.lat * (math.Pi / 180)) * math.Cos(latTo * (math.Pi / 180)) *		math.Sin(deltaLon / 2) * math.Sin(deltaLon / 2)	var c = 2 * math.Atan2(math.Sqrt(a),math.Sqrt(1-a))		return earthRadius * c}

在简单的二维平面上也是如此：

type CartesianPlaneData struct {	x, y float64}func (d CartesianPlaneData) calculateDistance(xTo, yTo float64) float64 {	// Simple 2-dimensional Euclidean distance 	dx := (xTo - d.x)	dy := (yTo - d.y)	return math.Sqrt( dx*dx + dy*dy )}

在这种情况下，main（）函数仅计算两种完全不同的距离类型：

func main() {	atlanta := GeospatialData{33.753746, -84.386330}	distance := atlanta.calculateDistance(33.957409, -83.376801) // to Athens, GA	fmt.Printf("The Haversine distance between Atlanta, GA and Athens, GA is %v\n", distance)	pointA := CartesianPlaneData{1, 1}	distanceA := pointA.calculateDistance(5, 5) 	fmt.Printf("The Pythagorean distance from (1,1) to (5,5) is %v\n", distanceA)}

当然，这是一个非常简单的示例，仅将结构强制应用到您的结构上。 但是，与Python相比，我的选择受到限制：

class CoordinateData:    def calculateDistance(self, latTo, lonTo):        pass class GeospatialData(CoordinateData):    def __init__(self, lat, lon):        self.lat = lat        self.long = lon    def calculateDistance(self, latTo, lonTo):        # Haversine goes here :)        return 95.93196816811724class CartesianPlaneData(CoordinateData):    def __init__(self, x, y):        self.x = y        self.x = y    # Let's not implement calculateDistance()if __name__ == "__main__":    atlanta = GeospatialData(33.753746, -84.386330)    distance = atlanta.calculateDistance(33.957409, -83.376801) # to Athens, GA    print('The Haversine distance between Atlanta, GA and Athens, GA is {}'.format(distance))    pointA = CartesianPlaneData(1,1)    distanceA = pointA.calculateDistance(5, 5)    print('The Pythagorean distance from (1,1) to (5,5) is {}'.format(distanceA))    print('pointA is of type {}'.format(pointA.__class__.__bases__))

这是有效的Python-CartesianPlaneData是CoordinateData的子类（而不是接口-Python使用Duck类），因此，仅使用不带返回类型（参见上面的静态类型与动态类型）的calculateDistance方法即可运行并返回：

python3 interface_example.py
The Haversine distance between Atlanta, GA and Athens, GA is 95.93196816811724
The Pythagorean distance from (1,1) to (5,5) is None
pointA is of type (<class '__main__.CoordinateData'>,)

Python仍然允许接口抽象，并且您绝对可以使用类来定义层次结构和逻辑，正如我对下面的PluginInterceptor所做的那样，以确定自定义插件是否是已定义基类的一部分； 但是，在运行时不会强制执行此操作，如果执行不正确，可能会令人瞠目结舌。

class PluginInterceptor:    """Loads all allowed plugins, when they are a subclass of `BasePlugin` and have the constant `name` set (not `__name__`)    """    def __init__(self):        self.cls = BasePlugin        self.allowed_plugins = self.__load__allowed_plugins__()            def __get_all_subclasses__(self, cls):        return set(cls.__subclasses__()).union(            )         def __load__allowed_plugins__(self):        __allowed_plugins__ = {}        for cls in self.__get_all_subclasses__(self.cls):            if cls.name:                __allowed_plugins__ = cls        return __allowed_plugins__

https://github.com/otter-in-a-suit/scarecrow/blob/master/plugin_base/interceptor.py#L5

package mainimport "fmt"func main() {	i, j := 42, 2701	p := &i         // point to i	fmt.Println(*p) // read i through the pointer	*p = 21         // set i through the pointer	fmt.Println(i)  // see the new value of i	p = &j         // point to j	*p = *p / 37   // divide j through the pointer	fmt.Println(j) // see the new value of j}

import numpy as npimport matplotlib.pyplot as plt# counts the number of iterations until the function diverges or# returns the iteration threshold that we check untildef countIterationsUntilDivergent(c, threshold):    z = complex(0, 0)    for iteration in range(threshold):        z = (z*z) + c        if abs(z) > 4:            break            pass        pass    return iteration

def mandelbrot(threshold, density):    # location and size of the atlas rectangle    # realAxis = np.linspace(-2.25, 0.75, density)    # imaginaryAxis = np.linspace(-1.5, 1.5, density)    realAxis = np.linspace(-0.22, -0.219, 1000)    imaginaryAxis = np.linspace(-0.70, -0.699, 1000)    realAxisLen = len(realAxis)    imaginaryAxisLen = len(imaginaryAxis)    # 2-D array to represent mandelbrot atlas    atlas = np.empty((realAxisLen, imaginaryAxisLen))    print('realAxisLen: {}, imaginaryAxisLen: {}'.format(realAxisLen, imaginaryAxisLen))        # color each point in the atlas depending on the iteration count    for ix in range(realAxisLen):        for iy in range(imaginaryAxisLen):            cx = realAxis            cy = imaginaryAxis            c = complex(cx, cy)            atlas = countIterationsUntilDivergent(c, threshold)            pass        pass    return atlas.T

import multiprocessing as mpimport itertoolsdef calc_row(cx, cy, threshold=120):    c = complex(cx<1>, cy<1>)    return (cx<0>, cy<0>, countIterationsUntilDivergent(c, threshold))

def mandelbrot_multi(threshold, density, cpus=4):    realAxis = np.linspace(-0.22, -0.219, 1000)    imaginaryAxis = np.linspace(-0.70, -0.699, 1000)    realAxisLen = len(realAxis)    imaginaryAxisLen = len(imaginaryAxis)    atlas = np.empty((realAxisLen, imaginaryAxisLen))        # Create list of permutations    realAxis = <(i,e ) for i,e in enumerate(realAxis)>     imaginaryAxis = <(i,e ) for i,e in enumerate(imaginaryAxis)>     paramlist = list(itertools.product(realAxis, imaginaryAxis))    paramlist = list(map(lambda t: t + (threshold,),paramlist))        # Create a multiprocessing pool    pool = mp.Pool(cpus)        n = pool.starmap(calc_row, paramlist)    pool.close()    pool.join()    return n, atlas

package mainimport (	"bytes"	"fmt"	"log"	"math/cmplx"	"os"	"strings"	"time"	"encoding/binary")func linspace(start, end float64, num int) <>float64 {	result := make(<>float64, num)	step := (end - start) / float64(num-1)	for i := range result {		result = start + float64(i)*step	}	return result}

func countIterationsUntilDivergent(c complex128, threshold int64) int64 {	z := complex(0, 0)	var ix int64 = 0	for i := int64(0); i < threshold; i++ {		ix = i		z = (z * z) + c		if cmplx.Abs(z) > 4 {			return i		}	}	return ix}func mandelbrot(threshold, density int64) <><>int64 {	realAxis := linspace(-0.22, -0.219, 1000)	imaginaryAxis := linspace(-0.70, -0.699, 1000)	fmt.Printf("realAxis %v\n", len(realAxis))	fmt.Printf("imaginaryAxis %v\n", len(imaginaryAxis))	atlas := make(<><>int64, len(realAxis))	for i := range atlas {		atlas = make(<>int64, len(imaginaryAxis))	}	fmt.Printf("atlas %v\n", len(atlas))	for ix, _ := range realAxis {		for iy, _ := range imaginaryAxis {			cx := realAxis			cy := imaginaryAxis			c := complex(cx, cy)			//fmt.Printf("ix, iy: %v %v\n", ix, iy)			atlas = countIterationsUntilDivergent(c, threshold)		}	}	return atlas}

type triple struct {	ix, iy int64	c      int64}func calcRow(ix, iy int64, c complex128, threshold int64) triple {	return triple{ix, iy, countIterationsUntilDivergent(c, threshold)}}

func mandelbrot(threshold, density int64) <><>int64 {	realAxis := linspace(-0.22, -0.219, 1000)	imaginaryAxis := linspace(-0.70, -0.699, 1000)	atlas := make(<><>int64, len(realAxis))	for i := range atlas {		atlas = make(<>int64, len(imaginaryAxis))	}	// Make a buffered channel	ch := make(chan triple, int64(len(realAxis))*int64(len(imaginaryAxis)))	for ix, _ := range realAxis {		go func(ix int) {			for iy, _ := range imaginaryAxis {				cx := realAxis				cy := imaginaryAxis				c := complex(cx, cy)				res := calcRow(int64(ix), int64(iy), c, threshold)				ch <- res			}		}(ix)	}	for i := int64(0); i < int64(len(realAxis))*int64(len(imaginaryAxis)); i++ {		select {		case res := <-ch:			atlas = res.c		}	}	return atlas}

class CountWords(beam.PTransform):  def expand(self, pcoll):    return (        pcoll        # Convert lines of text into individual words.        | 'ExtractWords' >>        beam.FlatMap(lambda x: re.findall(r'+', x))        # Count the number of times each word occurs.        | beam.combiners.Count.PerElement())counts = lines | CountWords()

func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {	s = s.Scope("CountWords")	// Convert lines of text into individual words.	col := beam.ParDo(s, extractFn, lines)	// Count the number of times each word occurs.	return stats.Count(s, col)}