dbf 文件批量自动同步

1.1 应用场景

希望将二进制的 dbf 文件处理成二维表并定时自动同步，可使用本文方案，为企业降本增效。

1.2 实现思路

• 在FDL服务器中按日期新建存放 dbf 文件到文件夹；

• 将业务数据系统中的 dbf 文件同步到FDL服务器下的指定文件夹；

• 使用 Shell 脚本遍历文件传输的文件名文件，生成 csv 文件；

• 使用 Python 算子中 dbf read 模块读取 dbf 文件。

本文用到的脚本文件：bdf.zip

2.1 新建文件夹

需要把在业务服务器中的 dbf 文件传输到 FDL 服务器中，以便使用 python 算子读取处理文件。因此要在 FDL 服务器先创建一个按照日期存储 dbf 文件的文件夹。

新建与 FDL 服务器的数据连接，详情参见：SSH 协议数据连接

使用 Shell 脚本，创建一个按当前执行日期命名的文件夹，用于存放 dbf 文件等。

其中  Shell 脚本的内容如下所示：

#!/bin/bash

# 
now=$1

# 创建文件夹
# 文件地址写服务器有权限的目录
mkdir -p "/data/demo_files/$now"

# 输出提示信息
echo "文件夹 '$now' 已在/data/demo_files/$now 路径下创建。"

注：需要根据实际情况修改目录路径。

将脚本放入 FDL服务器指定位置，并在 FineDataLink 中设置一个新的时间参数，如下图所示：

然后设置 Shell 脚本，输入脚本所在服务器路径和脚本使用的参数，如下图所示：

运行 Shell 脚本后，即可在服务器中看到新建的按运行日期显示的日期文件夹，如下图所示：

2.2 将 dbf 文件传输到指定文件夹

需要把在业务服务器中的 dbf 文件传输到 FDL 服务器中，以便使用 python 算子读取处理文件。

创建 配置FTP/SFTP数据连接，连接 dbf 文件所在服务器以及 FDL 服务器后台，分别作为文件来源和文件去向。

设置文件去向，数据源选择 FDL 服务器，文件夹地址选择 2.1 节创建的日期文件夹，如下图所示：

运行脚本后，即可在 FDL 服务器后台看到传输的文件，如下图所示：

2.3 生成包含文件名的 csv 文件

使用 Shell 脚本遍历文件传输的文件名文件，生成 csv 文件。

将脚本放置在FDL 服务器后台，脚本如下所示：

#!/bin/bash

now=$1

# 设置目标文件夹
target_folder="/data/demo_files/$now"

# 确保传入的参数不为空
if [ -z "$now" ]; then
    echo "请提供学生姓名作为参数。"
    exit 1
fi

# 设置输出文件名，包含学生姓名
output_file="$target_folder/file_list.csv"

# 清空输出文件
echo "Filename" > "$output_file"

# 检查目标文件夹是否存在
if [ -d "$target_folder" ]; then
    # 遍历目标文件夹中的所有 .dbf 文件
    for file in "$target_folder"/*.dbf; do
        if [ -f "$file" ]; then
            # 将文件名（不含路径）写入输出文件
            basename "$file" >> "$output_file"
        fi
    done
    echo "文件名已成功输出到 $output_file"
else
    echo "指定的文件夹 $target_folder 不存在。"
fi
显示代码

 SSH 连接选择 FDL 服务器后台（存放 2.2 节传输的 dbf 文件位置），脚本路径选择存放脚本的位置，并使用脚本参数，如下图所示：

运行脚本后会新增文件，记录所有的 dbf 文件名，如下图所示：

2.4 读取 dbf 文件并输出至指定数据库

使用 Python 算子中 dbf read 模块读取 dbf 文件。

使用「Python 算子」，需要准备 Python 环境，详情参见：Python算子环境准备

首先将 dbf 文件名设置为参数，以便后续使用算子进行批量读取。使用参数赋值，数据源选择 2.3 节 csv 文件所在的服务器，设置文件地址为 csv 所在的文件地址，如下图所示：

输出参数为 filename，如下图所示：

使用循环容器，并将参数 filename 设置为遍历对象，如下图所示：

使用数据转换节点并将其拖入循环容器中，进入数据转换编辑界面，如下图所示：

使用 Python 算子，输入代码如下：

import pandas as pd
from dbfread import DBF

# 设置 DBF 文件路径
dbf_path =  r'/opt/fdl/ceshi/${now}/${filename}'

table = DBF(dbf_path, encoding='ascii')
df = pd.DataFrame(iter(table))
output = df

使用 DB 输出算子，将读取的二维数据写入至指定的数据库中，如下图所示：

2.5 效果查看

运行任务后，即可看到运行时间当天的所有 dbf 文件数据被写入，如下图所示：