Day21 NiFi - 与 GCP BigQuery 对接设定

今天要来介绍如何透过 NiFi 来与 GCP BigQuery 来做资料整合与操作。

What is the GCP BigQuery?

在一开始先简单来为各位介绍，何谓 GCP BigQuery? BigQuery 是 GCP 中一个可以提供大量资料储存与查询分析的服务，我们可以透过 SQL 的方式来做一些资料的操作(ex. JOIN, GROUPBY 等)。

简单来说可以想像成他是个 DataWarehouse，在 BigQuery 中他会以 Column-Based 的方式来做资料的储存，再加上其对於资料老屈与分析时的效能能够提供到一个即时的程度，官方称 Near Realtime。

所以在应用上，我们可想像成无论是 Batch 或 Streaming 的资料，都可以进一步地有效写入，好让後续的使用者可以在该服务上面做立即性地查询。因此在 NiFi 这边就有提供了两个相关的 Processors，分别是 PutBigQueryBatch 和 PutBigQueryStreaming。

再来，我们从架构上来看一下 BigQuery 的资料阶层：

每一个 Project 底下会有多个 Datasets，而每一个 Datasets 则会有多个 Tables，我们就可以将这些 Tables 做整合应用与分析。

How to use?

这边你可能会思考，为什麽在 NiFi 中 BigQuery 的 Processors 只支援写入而没有读取呢？其实一样有支援读取的 Processor，他的设定就跟我们之前在 AWS 捞取 Athena 和 Redshift 的操作一模一样，就是下载 BigQuery 的 JDBC Driver，然後指定好 Controller Service，就可以透过 SQL 相关的 Processor 做资料的读取与操作了，所以整体的设定上都是一样的概念。

如同前面提到的，这边 NiFi 本身提供了PutBigQueryBatch 和 PutBigQueryStreaming 这两个 Processor，从命名我们可以得知为一个差别在於 Batch 或 Streaming 的写入，所以在一些设定上也会有一些差异。

PutBigQueryBatch

该 Process 就是以 Batch 的方式来写入 FlowFiles 到 GCP BigQuery，相关设定如下：

• Project ID: GCP 的 Project ID
• GCP Credentials Provider Service: 先前设定好的 GCP Controller Service。
• Datasets: BigQuery Datasets 名称。
• Table Name: BigQuery Table Name 名称。
• Ignore Unknown Values: 是否要忽略未知的值，若选择是，则有异常的 record 则直接略过。

其他比较重要的设定是 CSV 相关的参数，如果要使用这个 Processor 的时候，必须确保好你的 FlowFiles 的 Content 为 CSV 格式，可以透过 CSVWriter 来做转换，所以原则上你在执行该 Processor 之前的 Content 可能会长得以下这样：

PassengerId,Pclass,Name,Sex,Age,SibSp,Parch,Ticket,Fare,Cabin,Embarked
892,3,"Kelly, Mr. James",male,34.5,0,0,330911,7.8292,,Q
893,3,"Wilkes, Mrs. James (Ellen Needs)",female,47,1,0,363272,7,,S
894,2,"Myles, Mr. Thomas Francis",male,62,0,0,240276,9.6875,,Q
895,3,"Wirz, Mr. Albert",male,27,0,0,315154,8.6625,,S
896,3,"Hirvonen, Mrs. Alexander (Helga E Lindqvist)",female,22,1,1,3101298,12.2875,,S
897,3,"Svensson, Mr. Johan Cervin",male,14,0,0,7538,9.225,,S
898,3,"Connolly, Miss. Kate",female,30,0,0,330972,7.6292,,Q
899,2,"Caldwell, Mr. Albert Francis",male,26,1,1,248738,29,,S
900,3,"Abrahim, Mrs. Joseph (Sophie Halaut Easu)",female,18,0,0,2657,7.2292,,C
901,3,"Davies, Mr. John Samuel",male,21,2,0,A/4 48871,24.15,,S
902,3,"Ilieff, Mr. Ylio",male,,0,0,349220,7.8958,,S
903,1,"Jones, Mr. Charles Cresson",male,46,0,0,694,26,,S

然而你每一次要 batch 的 Size 取决於你 FlowFiles 的 Content 有多少 record，所以你会发现在 Content 必须是 CSV 的格式下，PutBigQueryBatch 就会有许多关於 CSV 设定的参数要去注意。

PutBigQueryStreaming

这个 Processor 就是针对 Streaming 来做使用的，所以通常用於一个 FlowFiles 只有一笔 Record Content，然而当中的设定较少，但大多数都与 PutBigQueryBatch 大同小异，一样列在下面给各位做参考：

• Project ID: GCP 的 Project ID
• GCP Credentials Provider Service: 先前设定好的 GCP Controller Service。
• Datasets: BigQuery Datasets 名称。
• Table Name: BigQuery Table Name 名称。
• Ignore Unknown Values: 是否要忽略未知的值，若选择是，则有异常的 record 则直接略过。
• Record Reader: 这是唯一一个不一样的参数，我们可以讲多种类型的 Content 写入到 BigQuery，透过指定好 JsonReader, CsvReader, ParquetReader等，所以相较於PutBigQueryBatch 则来得更有弹性。

小总结

介绍完了 GCP BigQuery 如何设定之後，不难发现他的设定与参数原理也十分单纯，只要掌握好设定原则之後，接着与上下游 Processor 做整合之後，你也可以将你要的资料读取或写入到 GCP BigQuery。

明天是最後一天介绍 GCP，我会带大家来理解 GCP PubSub 是如何在 NiFi 做设定与整合，如此一来你也可以透过 GCP 的 Message Queue 系统来与 NiFi 做整合，来形成一个更完整的 Data Pipeline 系统。

Reference

• Apache NiFi Document