Data Fusion 中 SFTP 插件的文件路径过滤器

Question

我正在使用 GC Data Fusion 中的 SFTP 插件从支付网关操作员下载财务报告，这些报告以 csv 文件形式存储在服务器上。

我试图避免上传所有文件以节省一些数据处理成本，结果发现该插件提供了类似“文件路径过滤器”的功能，如下所示：

文档中没有关于如何使用此属性的具体示例，而且我一直在尝试的任何方法都不起作用

文件的存储和命名如下：

SettlementInvoice GS Companyname Limited 190024000478 20220315.csv

所以我的假设是我可以使用：

/20220315/g

然后我可以将 20220315 的宏与来自另一个源的当前日期一起使用（插件支持宏）。

但是，经过测试后它不起作用，并且插件会忽略我在过滤器中输入的任何内容。这可能是由于语法造成的，但我找不到任何如何正确使用此选项的示例。如果有任何关于此事的建议，我将不胜感激。

Answer 1

我能够找到可用于过滤文件的模式。该模式是使用规则 我在评论中提到过。

我使用下面的 Java 程序来测试该模式，然后使用 来自 CDAP Hub 的 SFTP 操作插件。该管道将​​文件从 SFTP 源复制到 GCS。

import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class sftpCopyPluginRegex {
    public static void main(String[] args) {
        
        Pattern p = Pattern.compile("[a-zA-Z0-9 ]*20220320[.a-z]*");

        /*
            Explanation: 
            [a-zA-Z0-9 ]* => Allows lower-case and upper-case alphabets and numbers with zero or more instances
            20220320 => File name to match
            [.a-z]* => Allows lower-case alphabets and "." punctuation with zero or more instances
        */ 
        
        Matcher m = p.matcher("SettlementInvoice GS Companyname Limited 190024000478 20220320.csv"); // File name to match
        // Regex with macros ⇒ [a-zA-Z0-9 ]*${file.name}[.a-z]*

        boolean b = m.matches();
        System.out.println(b);

    }
}

使用的正则表达式是 [a-zA-Z0-9 ]*${file.name}[.az]* ，它根据中提供的日期过滤文件宏。数据融合管道 JSON 可以在下面找到。要测试此管道，您可以导入 JSON 文件并根据需要编辑文件系统属性。
Data Fusion 实例版本为 6.4.1。

{
    "artifact": {
        "name": "cdap-data-pipeline",
        "version": "6.4.1",
        "scope": "SYSTEM"
    },
    "description": "Data Pipeline Application",
    "name": "sftp-to-gcs_v1",
    "config": {
        "resources": {
            "memoryMB": 2048,
            "virtualCores": 1
        },
        "driverResources": {
            "memoryMB": 2048,
            "virtualCores": 1
        },
        "connections": [],
        "comments": [],
        "postActions": [],
        "properties": {},
        "processTimingEnabled": true,
        "stageLoggingEnabled": false,
        "stages": [
            {
                "name": "SFTPCopyTest",
                "plugin": {
                    "name": "SFTPCopy",
                    "type": "action",
                    "label": "SFTPCopyTest",
                    "artifact": {
                        "name": "sftp-actions",
                        "version": "1.5.1",
                        "scope": "USER"
                    },
                    "properties": {
                        "host": "<host-name>",
                        "port": "22",
                        "userName": "<user-name>",
                        "Authentication": "password-select",
                        "srcDirectory": "<sftp-source-directory>",
                        "destDirectory": "gs://<bucket-name>/<path-inside-the-bucket>",
                        "variableNameHoldingFileList": "sftp.copied.file.names",
                        "extractZipFiles": "false",
                        "password": "password",
                        "fileSystemProperties": "fs.gs.impl=>com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem,fs.AbstractFileSystem.gs.impl=>com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS,fs.gs.project.id=><project-id>,fs.gs.system.bucket=><bucket-name>,fs.gs.path.encoding=><path-inside-the-bucket>,fs.gs.working.dir=>/,fs.gs.impl.disable.cache=>true",
                        "fileNameRegex": "[a-zA-Z0-9 ]*${file.name}[.a-z]*"
                    }
                },
                "outputSchema": [
                    {
                        "name": "etlSchemaBody",
                        "schema": ""
                    }
                ],
                "id": "SFTPCopyTest"
            }
        ],
        "schedule": "0 * * * *",
        "engine": "spark",
        "numOfRecordsPreview": 100,
        "description": "Data Pipeline Application",
        "maxConcurrentRuns": 1
    }
}

注意：根据这个JIRA，建议使用操作插件而不是源插件，因为 SFTP/FTP 源插件将来会被删除。