将 Postgres 函数转换为 Impala UDF 或 Spark 中的函数

Posted 2023-03-22

【中文标题】将 Postgres 函数转换为 Impala UDF 或 Spark 中的函数

【英文标题】：Converting Postgres Function to Impala UDF or a function in Spark

【发布时间】：2016-07-13 16:14:23

【问题描述】：

我有一个在查询中调用的 postgres 函数。它类似于此示例：

CREATE OR REPLACE FUNCTION test_function(id integer, dt date, days int[], accts text[], flag boolean) RETURNS float[] AS $$
  DECLARE
    pt_dates date[];
    pt_amt integer[];
    amt float[];
  BEGIN
  if cleared then
      pt_dates := array(select dt from tabl);
      pt_amt := array(select amt from tab1);
      if array_upper(days, 1) is not null then
       for j in 1 .. array_upper(days, 1)
       loop
         amt+=amt;
       end loop;
      end if;
      return amt;
  END;
$$ LANGUAGE plpgsql;

如果我希望将其转换为 Data Lake 环境，最好的方法是什么？黑斑羚UDF？或火花 UDF ？还是蜂巢 UDF？在 Impala UDF 中，如何访问 impala 数据库？如果我编写 Spark UDF，我可以在 impala-shell 中使用它吗？

请指教。

【参考方案1】：

您的 1 个帖子中有很多很多问题。所以我只选择与 Spark 相关的问题。

您拥有表示您希望执行的数据处理的 SQL 查询。

以下是使用 Spark 执行此操作的通用公式：

获取一些数据，将其移至 S3 进入 AWS EMR 并创建一个新集群 SSH进入主节点，运行pyspark控制台 一旦启动，您可以通过rdd = sc.readText("s3://path/to/your/s3/buckets/") 读入您的 S3 数据 使用map 函数rdd2 = rdd.map(..add schema..) 对其应用架构 将rdd2 转换为dataframe 并将其存储为新变量。 rdd2DF = rdd2.toDF() 对此执行rdd2DF.registerTempTable('newTableName') 编写 SQL 查询并存储结果：output = sqlContext.sql("SELECT a,b,c FROM newTableName") 显示输出：output.show()

现在我知道这实在是太高了，无法具体回答你的问题，但我刚才所说的一切非常适合谷歌。

这是一个分离的计算和存储场景的示例，利用 EMR 与 Spark 和 SparkSQL 通过 SQL 查询处理大量数据。