将.txt文件拆分为元素

Question

因为我们不能直接从Json文件读取，所以我使用的是.txt文件。 好像有更多的元素被“，”分开。你知道吗

[
  {
    "Item_Identifier": "FDW58", 
    "Outlet_Size": "Medium"

  },
  {
    "Item_Identifier": "FDW14",
    "Outlet_Size": "Small"
  },
]

我想数一数元素的数目，这里我得到2。 问题是我不能将文本分割成以逗号“，”分隔的元素。 即使我把每一行都转换成json格式，我也能单独得到它。你知道吗

lines = p | 'receive_data' >> beam.io.ReadFromText(
    known_args.input)\
    | 'jsondumps' >> beam.Map(lambda x: json.dumps(x))\
    | 'jsonloads' >> beam.Map(lambda x: json.loads(x))\
    | 'print' >> beam.ParDo(PrintFn()) \

Answer 1

.json文件只是一个文本文件，其内容是JSON可解析格式的。你知道吗

JSON无效，因为它有尾随逗号。这应该起作用：

import json

j = r"""
[
    {
        "Item_Identifier": "FDW58",
        "Outlet_Size": "Medium"
    },
    {
        "Item_Identifier": "FDW14",
        "Outlet_Size": "Small"
    }
]
"""

print(json.loads(j))

Answer 2

我不认为这是一个安全的方法。我没有使用pythonsdk（我使用java），但是java端的io.TextIO非常清楚，它将发出一个PCollection，其中每个元素都是源文件的一行输入。分层数据格式（json、xml等）不适合以这种方式拆分。你知道吗

如果文件的格式和包含的json一样好，而且没有嵌套，那么您就可以不必担心：

• 逐行阅读文件（我相信你正在这样做）
• 仅筛选包含}的行
• 计算结果pcollection大小

不过，为了更广泛地与json集成，我们采用了不同的方法：

• 从字符串的PCollection开始，其中每个值都是文件的路径
• 使用本机库访问文件并以流式方式解析它（我们使用scala，它有一些可用的流式json解析库）
  • 或者，使用Beam的api从MatchResult获取ReadableFile实例并通过该实例访问文件

我的理解是，并不是所有的文件格式都适合分布式处理器。例如，Gzip不能很容易地被“分割”或分块。与JSON相同。CSV有一个问题，即值是无意义的，除非你手头上还有开头行。你知道吗