JSON解析前过滤出不合法的数据

1.1 应用场景

大数据量场景下的 json 数据解析，存在问题如下：

• 如果存在几条不合法的 json 格式数据，解析时会导致整个定时任务运行终止。

• json 解析是数据处理过程，定时任务的脏数据容忍没办法排除 json 不合法数据对任务的影响。

用户希望：

• 过滤出不合法的 json 格式数据，使其不影响定时任务的运行。

• 大数据量场景下，快速找出不合法的 json 格式数据。
  

1.2 实现思路

使用 Python 的 is_valid_json 判断 json 是否合法，合法的的 json 参与解析。

1.3 任务展示

FineDataLink 中的数据处理过程，详情参见：https://demo.finedatalink.com/ 「JSON解析前过滤出不合法的数据」。

2.1 准备工作

1）本文方案中需要用到「Python算子」，需参考 Python 算子 文档准备环境、了解使用方法。

2）本文方案中，json 数据保存在 txt 文件中，所以需要准备：配置FTP/SFTP数据连接、或者配置服务器本地目录数据连接

待解析的 json 数据如下所示：

json 示例数据：test_1.txt

2.2 读取 json 数据

1）新建定时任务，拖入「数据转换」节点，进入「数据转换」节点。

2）本文方案中，json 数据保存在 txt 文件中，所以需要拖入「文件输入」算子读取 json 数据。用户可根据实际情况修改该步骤。如下图所示：

「读取文件后缀」设置txt；列分隔符为「无」；不勾选「首行为字段名」，输出字段选择「手动获取」，输出字段类型为varchar、名称为column。

点击「数据预览」，如下图所示：

2.3 过滤出不合法json数据

1）拖入「Python算子」，使用 Python 的 is_valid_json 判断 json 是否合法。如下图所示：

注：下方代码中的 input 的值需要根据实际情况修改。

import pandas as pd
# 必须使用pandas库
# 若连接了数据源，可以鼠标点击上方相应数据源进行使用，输入源在python中以pandas的Dataframe数据类型存在，可以通过Dataframe的方法进行数据处理
# ----------------------------------------
import json  
  
def is_valid_json(json_string):  
    try:  
        json.loads(json_string)  
        return True  
    except json.JSONDecodeError:  
        return False  
  
# 使用示例  
input=$[文件输入]
a=[]
for row in input.index:
  json_string=input.loc[row]['column']
  a.append(is_valid_json(json_string))
input['isvalid']=a 

# ----------------------------------------
output = input
# 将需要输出给下游算子的数据赋值给output变量，当output是Dataframe数据类型时，输出为二维表的形式，当output是其他数据类型时，输出为字符串的形式

代码含义：

遍历输入数据的每一行，将每行的特定列（column列）中的 json 字符串进行解析，检查其是否是有效的 json 格式。然后，将检查结果添加到一个新列 isvalid 中，表示每个 json 字符串是否有效。最后，将带有新列的输入数据作为输出传递给下游算子。

2）拖入「数据过滤」算子，过滤出 isvalid 字段为真的数据（json 格式合法的数据）。如下图所示：

2.4 解析数据

1）拖入「JSON解析」算子，解析格式正确的数据。如下图所示：

2）点击「数据预览」，如下图所示：

2.5 输出数据

拖入「DB表输出」算子，将解析后的数据输出。写入方式选择「追加写入数据」（根据实际情况设置）。如下图所示：

2.6 效果查看

点击上方「运行」按钮，运行成功后，日志如下图所示：

数据落库后如下图所示：

2.7 后续步骤

可点击「发布」按钮，将定时任务发布到生产模式，如下图所示：

生产模式中，点击调度计划，可设置任务执行频率。