3

我有一个包含八个变量的 netCDF 文件。（对不起，不能分享实际文件）每个变量都有两个维度，时间和站点。时间大约是 14 步，车站目前有 38000 个不同的 id。因此，对于 38000 个不同的“位置”（实际上只是一个 id），我们有 8 个变量和 14 个不同的时间。

$ncdump -h stationdata.nc
netcdf stationdata {
dimensions:
    station = 38000 ;
    name_strlen = 40 ;
    time = UNLIMITED ; // (14 currently)
variables:
    int time(time) ;
            time:long_name = "time" ;
            time:units = "seconds since 1970-01-01" ;
    char station_name(station, name_strlen) ;
            station_name:long_name = "station_name" ;
            station_name:cf_role = "timeseries_id" ;
    float var1(time, station) ;
            var1:long_name = "Variable 1" ;
            var1:units = "m3/s" ;
    float var2(time, station) ;
            var2:long_name = "Variable 2" ;
            var2:units = "m3/s" ;
...

需要将此数据加载到 PostGres 数据库中，以便将数据连接到与 station_name 匹配的某些几何图形，以便以后进行可视化。

目前我已经在 Python 中使用 netCDF4 模块完成了这项工作。工作，但它需要永远！现在我像这样循环：

times = rootgrp.variables['time']
stations = rootgrp.variables['station_name']
for timeindex, time in enumerate(times):
    stations = rootgrp.variables['station_name']
    for stationindex, stationnamearr in enumerate(stations):
        var1val = var1[timeindex][stationindex]
        print "INSERT INTO ncdata (validtime, stationname, var1) \
            VALUES ('%s','%s', %s);" % \
            ( time, stationnamearr, var1val )

这需要在我的机器上运行几分钟，我觉得它可以以更聪明的方式完成。

任何人都知道如何以更智能的方式完成此操作？最好在 Python 中。

4

3 回答 3

3

不确定这是不是正确的方法，但我找到了解决这个问题的好方法，并认为我应该分享它。

在第一个版本中，脚本运行大约需要一个小时。重写代码后，它现在可以在不到 30 秒内运行！

最重要的是使用 numpy 数组并将来自 NetCDF 读取器的变量数组转换为行，然后将所有列堆叠到一个矩阵中。然后使用 psycopg2 copy_from 函数将该矩阵加载到数据库中。我从这个问题中得到了代码

使用带有 psycopg2 的二进制 COPY 表 FROM

我的部分代码：

dates = num2date(rootgrp.variables['time'][:],units=rootgrp.variables['time'].units)
var1=rootgrp.variables['var1']
var2=rootgrp.variables['var2']

cpy = cStringIO.StringIO()

for timeindex, time in enumerate(dates):

    validtimes=np.empty(var1[timeindex].size, dtype="object")
    validtimes.fill(time)

    #  Transponse and stack the arrays of parameters
    #    [a,a,a,a]        [[a,b,c],
    #    [b,b,b,b]  =>     [a,b,c],
    #    [c,c,c,c]         [a,b,c],
    #                      [a,b,c]]

    a = np.hstack((
              validtimes.reshape(validtimes.size,1),
              stationnames.reshape(stationnames.size,1),
              var1[timeindex].reshape(var1[timeindex].size,1),
              var2[timeindex].reshape(var2[timeindex].size,1)
    ))

    # Fill the cStringIO with text representation of the created array
    for row in a:
            cpy.write(row[0].strftime("%Y-%m-%d %H:%M")+'\t'+ row[1] +'\t' + '\t'.join([str(x) for x in row[2:]]) + '\n')


conn = psycopg2.connect("host=postgresserver dbname=nc user=user password=passwd")
curs = conn.cursor()

cpy.seek(0)
curs.copy_from(cpy, 'ncdata', columns=('validtime', 'stationname', 'var1', 'var2'))
conn.commit()
于 2013-04-26T08:41:42.747 回答
2

您可以进行一些简单的改进来加快速度。所有这些都是独立的，您可以尝试所有这些或只尝试几个，看看它是否足够快。它们的难度大致按升序排列：

  • 使用psycopg2数据库驱动，速度更快
  • 将整个插入块包装在事务中。如果您正在使用psycopg2，您已经在执行此操作 - 它会commit在最后自动打开您必须执行的事务。
  • 在一个数组中收集几行值，然后每 n 行执行一次多值 INSERT。
  • 使用多个连接通过辅助进程进行插入 - 请参阅multiprocessing模块。由于 GIL（全局解释器锁定）问题，线程无法正常工作。

如果您不想使用一个大事务，您可以设置synchronous_commit = off并设置 acommit_delay以便连接可以在磁盘刷新实际完成之前返回。如果您在一个事务中完成所有工作，这对您没有多大帮助。

多值插入

Psycopg2 不直接支持多值INSERT，但你可以写：

curs.execute("""
INSERT INTO blah(a,b) VALUES
(%s,%s),
(%s,%s),
(%s,%s),
(%s,%s),
(%s,%s);
""", parms);

并循环类似：

parms = []
rownum = 0
for x in input_data:
    parms.extend([x.firstvalue, x.secondvalue])
    rownum += 1
    if rownum % 5 == 0:
        curs.execute("""INSERT ...""", tuple(parms))
        del(parms[:])
于 2013-04-15T23:47:20.777 回答
2

组织您的循环以访问每次的所有变量。换句话说，一次读取和写入一条记录，而不是一次读取一个变量。这可以极大地加快速度，特别是如果源 netCDF 数据集存储在具有大磁盘块（例如 1MB 或更大）的文件系统上。有关为什么这更快的解释以及数量级导致的加速的讨论，请参阅此 NCO 加速讨论，从条目 7 开始。

于 2013-04-16T19:13:30.950 回答