标签: dask

我有一个这样的数据框：

df.head()
   day      time  resource_record  
0   27  00:00:00             AAAA  
1   27  00:00:00                A  
2   27  00:00:00             AAAA  
3   27  00:00:01                A  
4   27  00:00:02                A  

并想找出某些resource_records事物的存在次数。

我的第一个尝试是使用by返回的Series value_counts()，这看起来不错，但是由于以后没有drop()在中实现，因此我不允许以后再排除一些标签dask.Series。

因此，我尝试不打印不需要的标签：

for row in df.resource_record.value_counts().iteritems():
    if row[0] in ['AAAA']:
        continue
    print('\t{0}\t{1}'.format(row[1], row[0]))

哪个工作正常，但是如果我想进一步处理此数据并真的希望“清除”该怎么办。因此，我对文档进行了更多搜索并找到了mask()，但这也有些笨拙：

records = df.resource_record.mask(df.resource_record.map(lambda x: x in ['AAAA'])).value_counts()

我正在寻找一种方法，该方法将允许我仅对单个值进行计数，但count()对所有非NaN值进行计数。

然后我找到了str.contains()，但是我不知道如何处理使用以下代码返回的未记录的Scalar类型：

print(df.resource_record.str.contains('A').sum())

输出：

dd.Scalar<series-..., dtype=int64>

但是即使在查看了Scalar的代码之后，dask/dataframe/core.py我也没有找到获得其价值的方法。

您如何有效地计算数据框中某组值的出现？

这是对dash洗改数据的跟进问题。

我有一个现有的dask数据框df，希望在其中执行以下操作：

df['rand_index'] = np.random.permutation(len(df))

但是，这会产生错误Column assignment doesn't support type ndarray。我试图使用df.assign(rand_index = np.random.permutation(len(df))它给出相同的错误。

这是一个最小的（不是）工作示例：

import pandas as pd
import dask.dataframe as dd
import numpy as np

df = dd.from_pandas(pd.DataFrame({'A':[1,2,3]*10, 'B':[3,2,1]*10}), npartitions=10)
df['rand_index'] = np.random.permutation(len(df))

注意：

前面的问题提到了使用，df = df.map_partitions(add_random_column_to_pandas_dataframe, ...)但是我不确定这是否与该特定情况有关。

编辑1

我尝试过 df['rand_index'] = dd.from_array(np.random.permutation(len_df))，执行没有问题。当我检查时df.head()，似乎已经创建了新列。但是，当我看时df.tail()，rand_index是一堆NaNs。

实际上，只是为了确认我检查了df.rand_index.max().compute()哪个结果小于len(df)-1。所以这可能df.map_partitions是发挥作用的地方，因为我怀疑这是将dask分区的问题。在我的特定情况下，我有80个分区（不涉及示例情况）。

因此，我正在尝试读取大量包含水文数据的相对较大的netCDF文件。NetCDF文件全部如下所示：

<xarray.Dataset>
Dimensions:         (feature_id: 2729077, reference_time: 1, time: 1)
Coordinates:
  * time            (time) datetime64[ns] 1993-01-11T21:00:00
  * reference_time  (reference_time) datetime64[ns] 1993-01-01
  * feature_id      (feature_id) int32 101 179 181 183 185 843 845 847 849 ...
Data variables:
    streamflow      (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
    q_lateral       (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
    velocity        (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
    qSfcLatRunoff   (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
    qBucket         (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
    qBtmVertRunoff  (feature_id) float64 dask.array<shape=(2729077,), chunksize=(50000,)>
Attributes:
    featureType:                timeSeries
    proj4:                      +proj=longlat +datum=NAD83 +no_defs
    model_initialization_time:  1993-01-01_00:00:00
    station_dimension:          feature_id
    model_output_valid_time:    1993-01-11_21:00:00
    stream_order_output: …