将某些 Flink 操作优先于其他操作答案

【问题标题】：Prioritising Certain Flink Operations over Others将某些 Flink 操作优先于其他操作
【发布时间】：2018-05-15 11:45:20
【问题描述】：

我正在使用Flink v1.4.0。

我正在利用批处理 API 来执行一些 ETL，其中我有一个 DataSet<Employee >，其中 Employee 具有以下形式：

public class Employee implements Serializable {

private String name;
private double baseSalary;
private double bonus;
private double totalComp;

...
}

假设所有变量都有一个构造函数、setter 和 getter。

现在，我有一些操作正在应用，我认为是以顺序方式进行的，据此我正在尝试丰富DataSet<Employee> employees，如下所示：

...
DataSet<String> employees = env.fromCollection(employeesList);

DataSet<Employee> initEmployees = employees.map(new InitMapFunction());

DataSet<Employee> employeesEnrichedWithSalaryData = initEmployees.map(new SalaryMapFunction(salaryEnrichmentData));

DataSet<Employee> employeesEnrichedWithBonusData = employeesEnrichedWithSalaryData.map(new BonusMapFunction(bonusEnrichmentData));

DataSet<Employee> finalEmployeesData = employeesEnrichedWithSalaryData.map(new TotalCompMapFunction());
...

假设在我的包中某处我有以下MapFunction 实现：

final class InitMapFunction implements MapFunction<String, Employee>, Serializable {

    @Override
    public Employee map(String name) {

        Employee employee = Employee.newBuilder().build();
        employee.setName(name)
        return employee;
    }
}

final class SalaryMapFunction implements MapFunction<Employee, Employee>, Serializable {

    private Map<String, double> mapOfEmployeeVsSalary;

    SalaryMapFunction(Map<String, double> mapOfEmployeeVsSalary) {
        this.mapOfEmployeeVsSalary = mapOfEmployeeVsSalary;
    }

    @Override
    public Employee map(Employee employee) {

        if(mapOfEmployeeVsSalary.containsKey(employee.getName())) {
           employee.setSalary(mapOfEmployeeVsSalary.get(employee.getName()))
        }

        return employee;
    }
}

final class BonusMapFunction implements MapFunction<Employee, Employee>, Serializable {

    private Map<String, double> mapOfEmployeeVsBonus;

    SalaryMapFunction(Map<String, double> mapOfEmployeeVsBonus) {
        this.mapOfEmployeeVsBonus = mapOfEmployeeVsBonus;
    }

    @Override
    public Employee map(Employee employee) {

        if(mapOfEmployeeVsBonus.containsKey(employee.getName())) {
           employee.setBonus(mapOfEmployeeVsBonus.get(employee.getName()))
        }

        return employee;
    }
}

final class TotalCompMapFunction implements MapFunction<Employee, Employee>, Serializable {

    @Override
    public Employee map(Employee employee) {
        employee.setTotalComp(employee.getSalary + employee.getBonus);
        return employee;
    }
}

问题是：最终的DataSet (finalEmployeesData) 是否包含正确的值？我知道我可以一口气做完，但这不是这个问题的重点。我实现的代码的功能要求在不同的步骤中进行丰富。我已经确定了在处理数据集时未使用正确值丰富特定字段的情况，如上所述。我理解/怀疑这是由于延迟评估，并依赖于 Flink 进行的优化以计算最佳执行顺序（识别独立操作等）。这是正确的吗？

最后，我如何保证某个操作优先于另一个？如果将这些操作按如下方式链接在一起，输出是否会改变？

DataSet<Employee> finalEmployessData = env.fromCollection(employeesList)
                  .map(new InitMapFunction())
                  .map(new SalaryMapFunction(salaryEnrichmentData))
                  .map(new BonusMapFunction(bonusEnrichmentData))
                  .map(new TotalCompMapFunction());

【问题讨论】：

标签： java etl apache-flink

【解决方案1】：

Flink 不会改变操作的顺序。如果将程序定义为

DataSet<Y> result = input
  .map(new Map1())
  .map(new Map2())

那么Map2() 将始终应用于Map1() 的结果。

此外，无论您将函数一个一个地应用于不同的对象还是像上一个代码 sn-p 那样以流畅的风格应用这些函数，都没有区别。

您说过，您观察到某些值无法正确设置的情况。假设您运行的代码与此处显示的不完全相同，一个原因可能是 Flink 如何连接算子以及它如何在算子之间传送记录。在某些情况下（例如 map 函数的序列），Flink 通过方法调用传递记录以避免序列化成本。我们称之为函数链。链式函数融合到单个运算符中（例如，您可以在 Web UI 中看到）。显然，函数必须小心它们如何与接收和发出的对象交互。否则，同一记录可能会同时被两个函数修改。我建议仔细查看 Flink 文档中关于 object reusage 的部分。

【讨论】：