小编Mic*_*oni的帖子

我是工作调度员的新手,正在寻找一个在大数据集群上运行工作的人.我对可用的选择很困惑.发现Oozie与TWS,Autosys等现有产品相比有很多局限性.

在Oozie vs. Airflow上需要一些比较点.

感谢您的帮助.

我正在尝试使用Beam管道,以便将SequenceMatcher函数应用于大量单词.我(希望)除了WriteToText部分之外已经找到了所有的东西.

我已经定义了一个自定义ParDo(在此称为ProcessDataDoFn),它接受main_input和side_input,处理它们并输出像这样的字典

{u'key': (u'string', float)}

我的管道非常简单

class ProcessDataDoFn(beam.DoFn):
    def process(self, element, side_input):

    ... Series of operations ...

    return output_dictionary

with beam.Pipeline(options=options) as p:

    # Main input
    main_input = p | 'ReadMainInput' >> beam.io.Read(
        beam.io.BigQuerySource(
            query=CUSTOM_SQL,
            use_standard_sql=True
        ))

    # Side input
    side_input = p | 'ReadSideInput' >> beam.io.Read(
        beam.io.BigQuerySource(
            project=PROJECT_ID,
            dataset=DATASET,
            table=TABLE
        ))

    output = (
        main_input
        | 'ProcessData' >> beam.ParDo(
            ProcessDataDoFn(),
            side_input=beam.pvalue.AsList(side_input))
        | 'WriteOutput' >> beam.io.WriteToText(GCS_BUCKET)
    )

现在的问题是,如果我像这样离开管道,它只输出output_dictionary的键.如果我将ProcessDataDoFn的返回值更改为json.dumps(ouput_dictionary),则Json写得正确但是像这样

{
'
k
e
y
'

:

[
'
s
t
r …

我一直在尝试编写一个函数，将Google Cloud Storage存储桶中的多个文件加载到单个Pandas Dataframe中，但是我似乎无法使其正常工作。

import pandas as pd
from google.datalab import storage
from io import BytesIO


def gcs_loader(bucket_name, prefix):
  bucket = storage.Bucket(bucket_name)
  df = pd.DataFrame()
  for shard in bucket.objects(prefix=prefix):
    fp = shard.uri
    %gcs read -o $fp -v tmp
    df.append(read_csv(BytesIO(tmp))
  return df

当我尝试运行时说：

在命令行中引用的未定义变量：$ fp