1.1 应用场景

某国有企业旗下有 100 多家分公司，子公司财务每月需要向总部提交各种财务预算和决算表，如资产负债表、利润表、现金流量表等不同类型的数据表。总部财务需要手动对相同类型的财务报表进行数据汇总，并将其整理后向领导做出统一汇报。

现存问题：

• 总部财务手动汇总数据耗时较长，效率低下。

• 汇总的数据出问题，异常数据无法快速追踪。

本文提供方案，可自动化读取文件，为企业降本增效。

1.2 实现思路

• 通过执行 Shell 脚本文件，扫描指定文件夹下的 CSV/Excel 文件，输出包含扫描文件绝对路径信息的文件。

• 截取输出文件中的绝对路径，获得相对路径，并将相对路径作为参数输出。

• 使用相对路径循环读取要扫描的文件，并将文件数据输出到数据库表中。

• 定时扫描文件，将数据汇总到数据库表中。

注1：本文方案支持从 FTP/SFTP 服务器上读取 Excel 和 CSV 文件数据。

注2：本文方案使用到 Shell 脚本 节点，所以 FDL 工程需要在 Linux 系统下。

1.3 任务展示

FineDataLink 中的数据处理过程，详情参见：https://demo.finedatalink.com/ 「场景案例>数据开发场景>批量读取文件」。

本文示例是从 FineDataLink 服务器上，批量读取 CSV 文件数据并汇总到数据库表中。

2.1 场景模拟

某公司每月会定时上传文件到  csv  文件夹中，这些文件数据最终定时手工汇总到数据库中的某张表中，存在效率低下和容易出错的问题。

本文示例中，csv 文件夹包含两个 csv 文件。文件数据如下图所示：

2.2 方案说明

1）使用「Shell 脚本」节点执行 .sh 脚本文件，扫描指定文件夹下的 CSV/Excel 文件，输出包含扫描文件绝对路径信息的文件。

2）使用「Spark SQL」算子截取扫描文件的绝对路径，得到相对路径，「参数输出」算子将相对路径作为参数输出。

3）将待扫描文件的相对路径参数赋值给「循环容器」节点，循环容器节点内使用「文件输入」算子循环读取要扫描的文件，「DB表输出」算子将要扫描的文件数据输出到数据库表中。

4）设置定时调度，定时扫描文件，将数据汇总到数据库表中。

2.3 准备工作

本文使用「Shell 脚本」节点，扫描 FDL 服务器中指定文件夹下的 csv 文件，所以需要做以下准备。

1）由于需要在目标服务器执行指定 Shell 脚本，因此需要首先通过 SSH 远程连接至目标服务器。

因此新建 SSH 协议数据连接，参考：前提条件

2）在 /data/demo_FR/webroot/WEB-INF/local_files/ 目录下新建 df_files 文件夹，每月定时将包含待扫描文件的 csv 文件夹上传到 df_files 文件夹中。如下图所示：

注：由于要 新建服务器本地目录数据连接，从  FineDataLink 本地服务器上读取 CSV/Excel 文件数据，所以存放待扫描文件的文件夹需要在%FDL_HOME%/webroot/WEB-INF/local_files/下。

3）新建一个服务器本地目录数据连接，服务器本地目录选择存放 csv 文件夹的路径。

4）本文使用「Shell 脚本」节点，扫描指定文件夹下的 csv 文件，所以需要准备 .sh 文件。

新建 file_search.sh 文件。代码如下所示：

本文示例 file_search.sh 文件：file_search.zip

#!/bin/bash
# 获取输⼊参数
path=$1 # 指定要扫描的文件夹路径
output=$2 # 指定输出文件的路径
# 输出表头
echo "文件名,绝对路径,修改时间" > $output
# 遍历文件夹
for file in $(ls $path)
do
# 获取文件名、绝对路径和修改时间
filename=$(basename $file)
abspath=$(realpath $path/$file)
modtime=$(stat -c %y $path/$file | awk '{print $1,$2}')
# 将文件名、绝对路径和修改时间输出到csv文件
echo "$filename,$abspath,$modtime" >> $output
done
# 输出完成信息
echo "扫描完成！输出文件路径为：$output"

5）将 file_search.sh 文件上传到 csv 文件夹所在的目录中。如下图所示：

2.4 设置 Shell 脚本节点

本节目的：扫描指定文件夹下的 csv 文件，结果输出到 file_list.csv 表中。file_list.csv 表中包含 csv 文件夹下表信息的：文件名、绝对路径、修改时间。

1）新建定时任务，添加「Shell脚本」节点。

2）在「参数设置」中新增两个文本类型的参数。如下表所示：

3）「Shell脚本」节点中，「脚本路径」填写SSH目标端要执行的Shell脚本完整路径，「脚本参数」添加本节第二步设置的参数。如下图所示：

4）右键点击「Shell脚本」节点，点击运行节点，可生成  file_list.csv 表。如下图所示：

2.5 设置数据转换节点

本节目的：读取 2.4 节的输出文件 file_list.csv 表，截取扫描文件的绝对路径，得到相对路径，并将相对路径输出为参数。

2.5.1 读取输出文件

1）添加「数据转换」节点，与「Shell脚本」节点相连。

2）进入「数据转换」节点，添加「文件输入」算子，读取 file_list.csv 表。如下图所示：

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

2.5.2 获取扫描文件的相对路径

添加「Spark SQL」算子，获取扫描文件的相对路径。如下图所示：

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

2.5.3 将扫描文件的相对路径输出为参数

添加「参数输出」算子，将扫描文件的相对路径输出为参数。如下图所示：

2.6 设置循环

1）添加「循环容器」节点，设置如下图所示：

2）「循环容器」节点内添加「数据转换」节点，并进入「数据转换」节点内。

3）添加「文件输入」算子，通过待扫描文件的相对路径，读取要扫描的文件。如下图所示：

4）添加「DB表输出」算子，将扫描文件的数据汇总到某张表中。如下图所示：

5）保存并运行任务。

2.7 设置定时调度

每月末汇总一次数据。如下图所示：

test_1数据库中，huizong 表数据如下图所示：