使用数据流转换和导出 CDN 日志

 "httpRequest":{        "cacheHit":true,      "cacheLookup":true,      "remoteIp":"47.9.225.153",      "requestMethod":"GET",      "requestSize":"258",      "requestUrl":"http://35.186.123.123/sz2/v3/h_m/190116/v1WE7o/240p_6.m3u8",      "responseSize":"5663",      "status":200,      "userAgent":"Someagent/4.7.18_ww (Linux;Android 8.0.0) ExoPlayerLib/2.8.4"   },   "insertId":"12ufqm4g22a5lxc",   "jsonPayload":{        "@type":"type.googleapis.com/google.cloud.loadbalancing.type.LoadBalancerLogEntry",      "statusDetails":"response_from_cache"   },   "logName":"projects/your_project/logs/requests",   "receiveTimestamp":"2019-03-19T05:32:38.196703011Z",   "resource":{        "labels":{           "backend_service_name":"",         "forwarding_rule_name":"cdn-static-lb-forwarding-rule",         "project_id":"your_project",         "target_proxy_name":"cdn-static-lb-target-proxy",         "url_map_name":"cdn-static-lb",         "zone":"global"      },      "type":"http_load_balancer"   },   "severity":"INFO",   "spanId":"9450bbf396adb07b",   "timestamp":"2019-03-19T05:32:37.345092429Z",   "trace":"projects/your_project/traces/34078fcedc28912261bd05b66cd86f05"}

git clone https://github.com/GoogleCloudPlatform/DataflowTemplates

 public static PipelineResult run(Options options) {    // Create the pipeline    Pipeline pipeline = Pipeline.create(options);
    /*     * Steps:  *   1) Read string messages from PubSub     *   2) Window the messages into minute intervals specified by the executor.     *   3) Output the windowed files to GCS     */    pipeline        .apply("Read PubSub Events", PubsubIO.readStrings().fromTopic(options.getInputTopic()))        .apply("Transform", ParDo.of(new DoFn<String, String>(){          @ProcessElement          public void processElement(ProcessContext context) {            String message = context.element();            JSONObject mgsObj = new JSONObject(message);
            JSONObject httpReqObj = (JSONObject) mgsObj.get("httpRequest");            String remoteIp = httpReqObj.getString("remoteIp");            String url = httpReqObj.getString("requestUrl");            String status = new Integer(httpReqObj.getInt("status")).toString();            String timestamp = mgsObj.getString("timestamp");

            String transformedLog = "GCP, " + remoteIp + ", " + url + ", " + status  + ", " + timestamp;            context.output(transformedLog);          }        }))        .apply(            options.getWindowDuration() + " Window",            Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))
        // Apply windowed file writes. Use a NestedValueProvider because the filename        // policy requires a resourceId generated from the input value at runtime.        .apply(            "Write File(s)",            TextIO.write()                .withWindowedWrites()                .withNumShards(options.getNumShards())                .to(   new WindowedFilenamePolicy(                        options.getOutputDirectory(),                        options.getOutputFilenamePrefix(),                        options.getOutputShardTemplate(),                        options.getOutputFilenameSuffix()))                .withTempDirectory(NestedValueProvider.of(                    options.getOutputDirectory(),                    (SerializableFunction<String, ResourceId>) input ->                        FileBasedSink.convertToFileResourceIfPossible(input)))); 
    // Execute the pipeline and return the result.    return pipeline.run();  }

import org.apache.beam.sdk.transforms.DoFn;import org.apache.beam.sdk.transforms.ParDo;import org.json.JSONObject;import java.lang.String;import org.apache.beam.sdk.values.PCollection;

TEMPLATE_BUCKET=your_template_bucketLOG_OUTPUT_BUCKET=your_bucketPROJECT_ID=your_project
mvn compile exec:java \ -Dexec.mainClass=com.google.cloud.teleport.templates.TransformDumpLog \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=${PROJECT_ID} \ --stagingLocation=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/staging \ --tempLocation=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/temp \--templateLocation=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/templates/TransformDumpLog.json \ --runner=DataflowRunner \ --windowDuration=5m \ --numShards=1 \ --inputTopic=projects/${PROJECT_ID}/topics/pubsub_cdn_log \ --outputDirectory=gs://${LOG_OUTPUT_BUCKET}/cdn_log_export/ \ --outputFilenamePrefix=log \ --outputFilenameSuffix=.txt"--tempLocation=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/temp \--templateLocation=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/templates/TransformDumpLog.json \ --runner=DataflowRunner \ --windowDuration=5m \ --numShards=1 \ --inputTopic=projects/${PROJECT_ID}/topics/pubsub_cdn_log \ --outputDirectory=gs://${LOG_OUTPUT_BUCKET}/cdn_log_export/ \ --outputFilenamePrefix=log \ --outputFilenameSuffix=.txt"

gcloud dataflow jobs run log-dump \--gcs-location=gs://${TEMPLATE_BUCKET}/dataflow/pipelines/templates/TransformDumpLog.json \--region=asia-east1