好訊息，我正在使用批處理，而不是流式傳輸。我還是不明白這是怎么回事。顯然 pvalue.AsDict() 實際上并沒有給你一個字典值？只是一個 PCollect 的包裝。我怎么能把它變成一個 dict 以便我可以使用它？
這段代碼在init方法上失敗，就在我嘗試訪問操作時，就好像它是......字典一樣。
錯誤是
File "[path]", line 40, in process
location_indexes = [x[0] for x in self.operations["lookup_location_id"]]
TypeError: 'AsDict' object is not subscriptable [while running 'ParDo(Locations)']
這是罪魁禍首
class Locations(beam.DoFn): # Location_ID
def __init__(self, operations: dict):
self.locations: list = operations["lookup_location_id"]
def process(self, element: str):
# I have code here, not relevant
這就是我稱之為地點的地方......
locations = (
csv_data
| beam.ParDo(Locations(operations=beam.pvalue.AsDict(operations)))
| "Dedup locations" >> beam.Distinct()
)
操作是元組的集合。這是管道：
operations = (
transforms
| ParDo(Semantics(headers))
| GroupByKey()
headers 實際上是一個普通的 ol 串列。所以它作為 SideInput 效果很好。語意給我帶來了一些元組。
class Semantics(beam.DoFn):
def __init__(self, headers: list):
self.headers = headers
def process(self, element: list):
key = element[0]
value: list = [self.headers.index(element[2]), element[3]]
yield key, value
我還查看了除錯器中的 AsDict 操作物件。這是一團糟，我不知道我應該如何從中提取真正的價值。任何人都可以幫忙嗎？
uj5u.com熱心網友回復：
__init__DoFn 類的部分在創建時運行，因此您無法獲取從側輸入生成的資料。
您需要做的是將側輸入視圖傳遞給process方法。使用我自己的例子：
class ChangeCurrency(beam.DoFn):
def process(self, value, ratios):
current = value["currency"]
exchanged = {"Original": current}
for key in ratios[current]:
exchanged[key] = value["amount"] * ratios[current][key]
return [exchanged]
{..}
pipeline | ParDo(ChangeCurrency(), ratios=beam.pvalue.AsDict(rates_pc))
僅供參考，您不需要為此創建一個類，您可以使用一個簡單的函式：
def change_currency(value, ratios):
current = value["currency"]
exchanged = {"Original": current}
for key in ratios[current]:
exchanged[key] = value["amount"] * ratios[current][key]
return [exchanged]
{..}
pipeline | ParDo(change_currency, ratios=beam.pvalue.AsDict(rates_pc))
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/363835.html