【发布时间】：2022-12-17 03:09:55
【问题描述】：
我正在尝试读取存储在 GCS 存储桶中的 .tsv 文件的内容，并将每一行写入相应的 BigQuery 表。执行此操作时出现空指针异常，这可能是因为数据流作业中的 BigQueryIO.write() 步骤在使用 TextIO.read() 读取文件内容之前开始。
堆栈跟踪
Error message from worker: java.lang.NullPointerException: Cannot invoke gradle_inital_load.TableAndRow.getTab_name()" because "row" is null gradle_inital_load.ReadTarDir.getTableName(ReadTarDir.java:318) gradle_inital_load.ReadTarDir.lambda$0(ReadTarDir.java:287) org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$TableFunctionDestinations.getDestination(DynamicDestinationsHelpers.java:128) org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$TableFunctionDestinations.getDestination(DynamicDestinationsHelpers.java:114) org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1.processElement(PrepareWrite.java:71)下面是我的代码
package gradle_inital_load; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.SerializableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.Compression; import org.apache.beam.sdk.io.FileIO; import org.apache.beam.sdk.io.FileIO.ReadableFile; import org.apache.beam.sdk.io.FileSystems; import org.apache.beam.sdk.io.TextIO; import org.apache.beam.sdk.io.fs.MatchResult; import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions; import org.apache.beam.sdk.io.fs.ResourceId; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; import org.apache.beam.sdk.io.gcp.bigquery.TableDestination; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptions.DirectRunner; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.*; import org.apache.beam.sdk.transforms.DoFn.ProcessContext; import org.apache.beam.sdk.transforms.DoFn.ProcessElement; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.TypeDescriptors; import org.apache.beam.sdk.values.ValueInSingleWindow; import org.apache.commons.compress.archivers.tar.TarArchiveEntry; import org.apache.commons.compress.archivers.tar.TarArchiveInputStream; import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.api.services.bigquery.model.TableRow; import com.google.common.io.Files; import org.apache.beam.sdk.annotations.*; import java.io.BufferedReader; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStreamReader; import java.nio.channels.Channels; import java.util.ArrayList; import java.util.List; import org.apache.beam.runners.dataflow.DataflowRunner; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; public class ReadTarDir { private static final Logger LOG = LoggerFactory.getLogger(ReadTarDir.class); static String outputTableProject = "gpc-d-disc"; static String outputTableDataset = "user_events"; public static void main(String[] args) { // TODO Auto-generated method stub DataflowPipelineOptions dfOptions = PipelineOptionsFactory.as(MyOptions.class); dfOptions.setProject("gpc-d-disc"); dfOptions.setStagingLocation("gs://crs_user_events/staging"); dfOptions.setRegion("us-east4"); dfOptions.setTempLocation("gs://crs_user_events/temp"); dfOptions.setServiceAccount("data-sa@gpc-d-disc.iam.gserviceaccount.com"); dfOptions.setSubnetwork( "https://www.googleapis.com/compute/v1/projects/gpc-net-n-spoke-prj/regions/us-east4/subnetworks/n-subnet006-disc-use4"); dfOptions.setUsePublicIps(false); dfOptions.setRunner(DataflowRunner.class); DataflowRunner.fromOptions(dfOptions); Pipeline p = Pipeline.create(dfOptions); PCollectionView<String[]> filecontents = (PCollectionView<String[]>) p.apply("Read column headers", TextIO.read() .from("gs://crs_user_events/initial_load/column_headers.tsv").withCompression(Compression.AUTO)) .apply("Create column header Array", ParDo.of(new DoFn<String, String[]>(){ @ProcessElement public void processElement(ProcessContext c) { String fileLines = c.element().toString(); String[] Data = fileLines.split("\t"); c.output(Data); } })).apply(View.asSingleton()); PCollection<String> lines = p.apply("Read Files", TextIO.read().from("gs://crs_user_events/initial_load/test.tsv.gz").withCompression(Compression.GZIP)).setCoder(NullableCoder.of(StringUtf8Coder.of())); p.getCoderRegistry().registerCoderForClass(ReadTarDir.class, TableAndRowCoder.of()); PCollection<TableAndRow> tablerows = lines .apply("Transform File lines into TableAndRow", ParDo.of(new DoFn<String, TableAndRow>() { @ProcessElement public void processElement(ProcessContext c) { int tabnam_idx, indx; TableAndRow tbObj = null; String tabName = null; TableRow row =new TableRow(); //TableRow row = new TableRow(); String[] columns = c.sideInput(filecontents); String[] arr = c.element().split("\t"); if (arr.length > 0) { tabnam_idx = getIndex(columns, "channel"); indx = getIndex(columns, "page_event"); // ProductDetails if ((arr[tabnam_idx].toString()).contains("productdetails")) { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("detail_page_view_events_idl"); // tabName = String.format("%s:%s.%s", outputTableProject, // outputTableDataset,"Detail_Page_View_Events"); row.set("eventType", "detail-page-view"); int index = getIndex(columns, "evar6"); if (arr[getIndex(columns, "evar6")] != "") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } row.set("eventTime", arr[getIndex(columns, "date_time")]); row.set("experimentIds", arr[getIndex(columns, "evar104")]); row.set("productDetails.product.id", arr[getIndex(columns, "product_list")]); row.set("userInfo.userId", "1"); row.set("userInfo.ipAddress", arr[getIndex(columns, "ip")]); row.set("userInfo.userAgent", arr[getIndex(columns, "user_agent")]); row.set("userInfo.directUserRequest", "1"); row.set("uri", arr[getIndex(columns, "page_url")]); if (arr[getIndex(columns, "visit_referrer")] == "") { row.set("referrerUri", "1"); } else { row.set("referrerUri", arr[getIndex(columns, "visit_referrer")]); } } // Homepage if ((arr[tabnam_idx].toString()).contains("homepage1")) { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("home_page_view_events_idl"); // tabName = String.format("%s:%s.%s", outputTableProject, // outputTableDataset,"Home_Page_View_Events"); row.set("eventType", "home-page-view"); if (arr[getIndex(columns, "evar6")] != " ") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } } // Search indx = getIndex(columns, "page_event"); if ((arr[tabnam_idx].toString()).contains("search") && arr[indx] == "0") { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("search_events_idl"); // tabName = String.format("%s:%s.%s", outputTableProject, // outputTableDataset,"Pass Table Name here"); /* create row here */ row.set("eventType", "search"); if (arr[getIndex(columns, "evar6")] != " ") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } if (arr[getIndex(columns, "evar6")] != " ") { row.set("searchQuery", arr[getIndex(columns, "evar1")]); } else { row.set("searchQuery", arr[getIndex(columns, "evar2")]); } row.set("productDetails.product.id", arr[getIndex(columns, "product_list")]); } // Browse indx = getIndex(columns, "page_event"); if ((arr[tabnam_idx].toString()).contains("category-landing") && arr[indx] == "0") { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("category_page_view_events_idl"); /* create row here */ row.set("eventType", "category-page-view"); if (arr[getIndex(columns, "evar6")] != " ") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } row.set("pageCategories", arr[getIndex(columns, "evar104")]); } // add-to-cart if (arr[getIndex(columns, "product_list")] != null && arr[indx] == "12") { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("add_to_cart_events_idl"); /* create row here */ row.set("eventType", "add-to-cart"); if (arr[getIndex(columns, "evar6")] != " ") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } row.set("productDetails.product.id", arr[getIndex(columns, "product_list")]); } // purchase complete indx = getIndex(columns, "page_event"); if (arr[getIndex(columns, "product_list")] != null && arr[indx] == "1") { tabName = "outputTableProject".concat(":").concat(outputTableDataset).concat(".") .concat("purchase_complete_events_idl"); /* create row here */ row.set("eventType", "home-page-view"); if (arr[getIndex(columns, "evar6")] != " ") { row.set("visitorId", arr[getIndex(columns, "evar6")]); } else { row.set("visitorId", arr[getIndex(columns, "mcvisid")]); } row.set("productDetails.product.id", arr[getIndex(columns, "product_list")]); row.set("productDetails.product.quantity", arr[getIndex(columns, "product_list")]); row.set("purchaseTransaction.revenue", arr[getIndex(columns, "product_list")]); row.set("purchaseTransaction.currencyCode", arr[getIndex(columns, "product_list")]); } } LOG.info("Row:" + row.toString()); if(row!=null && tabName!=null) { tbObj = new TableAndRow(row, tabName); } c.output(tbObj); } }).withSideInputs(filecontents)).setCoder(NullableCoder.of(TableAndRowCoder.of())); tablerows.apply("Write to BigQuery", BigQueryIO.<TableAndRow>write().to(line -> getTableName(line)) .withFormatFunction((TableAndRow line) -> convertToTableRow(line)) .withCreateDisposition(CreateDisposition.CREATE_NEVER) .withWriteDisposition(WriteDisposition.WRITE_APPEND)); p.run().waitUntilFinish(); System.out.println("Pipeline Executed"); } private static TableRow convertToTableRow(TableAndRow line) { // TODO Auto-generated method stub TableRow row = line.getRow(); return row; } public static int getIndex(String[] Data, String str) { int index = -1; for (int j = 0; j < Data.length; j++) { if (Data[j].contains(str)) { index = j; break; } } return index; } public static TableDestination getTableName(ValueInSingleWindow<TableAndRow> line) { TableDestination destination = null; TableAndRow row = line.getValue(); if(row.getTab_name()!=null) { destination = new TableDestination(row.getTab_name(), null); } return destination; } }`
有人可以帮忙吗，因为我是 Dataflow Apache Beam 编程的新手。
应首先读取文件内容，并且必须将文件中的每一行转换为表行并返回到 BigQuery 表。表名也由文件中每一行的内容确定。
【问题讨论】：
标签： java google-bigquery pipeline google-cloud-dataflow apache-beam-io