Thiết kế website giá rẻ

Question

I have created a beam pipeline to read data from the Kafka topic and then insert it to hive tables in DataProc cluster

I consumed the data and converted it to the HcatRecord as below

p.getCoderRegistry()
        .registerCoderForClass(HCatRecord.class, WritableCoder.of(DefaultHCatRecord.class));


PCollection<KafkaRecord<byte[], byte[]>> collection = p.apply(KafkaIO.<byte[], byte[]>read()
                .withBootstrapServers("host:9092").withTopic("topic")
                .withConsumerConfigUpdates(ImmutableMap.of("specific.avro.reader", "true"))
                .withStartReadTime(Instant.now().minus(org.joda.time.Duration.standardDays(10)))
                .withKeyDeserializer(ByteArrayDeserializer.class)
                .commitOffsetsInFinalize()
                .withValueDeserializer(ByteArrayDeserializer.class)
                .withConsumerConfigUpdates(ImmutableMap.of("enable.auto.commit", "true"))
                .withConsumerConfigUpdates(ImmutableMap.of("group.id", "group"))

);

PCollection<String> deserializedRecords = collection.apply("DeserializeRecords", ParDo.of(new DeserializeDoFn("http://schema-registry:port")));

PCollection<HCatRecord> hCatRecords = deserializedRecords.apply(ParDo.of(new JsonToHCatRecordFn()));

Then I use HCatalogIO to insert data into table.

hCatRecords.apply(
        HCatalogIO.write()
                .withConfigProperties(configProperties)
                .withDatabase("database") //optional, assumes default if none specified
                .withTable("table")
                .withPartition(partition) //optional, may be specified if the table is partitioned
       

     .withBatchSize(1024L)
    );

when it’s runs on my local mchine it throws below exception

Exception in thread "main" org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.UncheckedExecutionException: org.apache.beam.sdk.util.UserCodeException: org.apache.hive.hcatalog.common.HCatException : 2004 : HCatOutputFormat not initialized, setOutput has to be called. Cause : org.apache.hive.hcatalog.common.HCatException : 2001 : Error setting output information. Cause : org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "gs"
    at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2086)
    at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.LocalCache.get(LocalCache.java:4012)
    at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4035)
    at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:5013)
    at org.apache.beam.runners.direct.DoFnLifecycleManager.get(DoFnLifecycleManager.java:64)
    at org.apache.beam.runners.direct.ParDoEvaluatorFactory.createEvaluator(ParDoEvaluatorFactory.java:131)
    at org.apache.beam.runners.direct.ParDoEvaluatorFactory.forApplication(ParDoEvaluatorFactory.java:81)
    at org.apache.beam.runners.direct.TransformEvaluatorRegistry.forApplication(TransformEvaluatorRegistry.java:158)
    at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:122)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.beam.sdk.util.UserCodeException: org.apache.hive.hcatalog.common.HCatException : 2004 : HCatOutputFormat not initialized, setOutput has to be called. Cause : org.apache.hive.hcatalog.common.HCatException : 2001 : Error setting output information. Cause : org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "gs"

Then i Tried to deploy it in the gcp dataflow so while the deployment process it throws below exception

java.lang.RuntimeException: Error while staging packages
    at org.apache.beam.runners.dataflow.util.PackageUtil.stageClasspathElements (PackageUtil.java:372)
    at org.apache.beam.runners.dataflow.util.PackageUtil.stageClasspathElements (PackageUtil.java:238)
    at org.apache.beam.runners.dataflow.util.GcsStager.stageFiles (GcsStager.java:53)
    at org.apache.beam.runners.dataflow.DataflowRunner.stageArtifacts (DataflowRunner.java:1068)
    at org.apache.beam.runners.dataflow.DataflowRunner.run (DataflowRunner.java:1217)
    at org.apache.beam.runners.dataflow.DataflowRunner.run (DataflowRunner.java:203)
    at org.apache.beam.sdk.Pipeline.run (Pipeline.java:325)
    at org.apache.beam.sdk.Pipeline.run (Pipeline.java:310)
    at org.pickme.rt.App.run (App.java:158)
    at org.pickme.rt.App.main (App.java:165)
    at jdk.internal.reflect.DirectMethodHandleAccessor.invoke (DirectMethodHandleAccessor.java:103)
    at java.lang.reflect.Method.invoke (Method.java:580)
    at org.codehaus.mojo.exec.ExecJavaMojo$1.run (ExecJavaMojo.java:282)
    at java.lang.Thread.run (Thread.java:1583)
Caused by: java.lang.NoClassDefFoundError: com/google/cloud/hadoop/util/TraceOperation
    at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.getObject (GoogleCloudStorageImpl.java:2327)
    at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.getItemInfo (GoogleCloudStorageImpl.java:2225)
    at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.getWriteGeneration (GoogleCloudStorageImpl.java:2295)
    at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.create (GoogleCloudStorageImpl.java:553)
    at org.apache.beam.sdk.extensions.gcp.util.GcsUtil.create (GcsUtil.java:619)
    at org.apache.beam.sdk.extensions.gcp.storage.GcsFileSystem.create (GcsFileSystem.java:154)
    at org.apache.beam.sdk.extensions.gcp.storage.GcsFileSystem.create (GcsFileSystem.java:71)
    at org.apache.beam.sdk.io.FileSystems.create (FileSystems.java:245)
    at org.apache.beam.runners.dataflow.util.PackageUtil.tryStagePackage (PackageUtil.java:215)
    at org.apache.beam.runners.dataflow.util.PackageUtil.tryStagePackageWithRetry (PackageUtil.java:174)
    at org.apache.beam.runners.dataflow.util.PackageUtil.stagePackageSynchronously (PackageUtil.java:153)
    at org.apache.beam.runners.dataflow.util.PackageUtil.lambda$stagePackage$1 (PackageUtil.java:142)
    at org.apache.beam.sdk.util.MoreFutures.lambda$supplyAsync$0 (MoreFutures.java:108)
    at java.util.concurrent.CompletableFuture$AsyncRun.run (CompletableFuture.java:1804)
    at java.util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor.java:1144)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run (ThreadPoolExecutor.java:642)
    at java.lang.Thread.run (Thread.java:1583)

Can you guys help me to figure out the issue here. also please share the documntes or guidence as well

Thiết kế website giá rẻ

Danh mục

Apache beam Pipeline | kafka to Dataproc