Design and create a reusable pipeline

Deploy the Cloud Storage Argument Setter plugin

1. In the Cloud Data Fusion web interface, go to the Studiopage.

2. In the Actionsmenu, click GCS Argument Setter.

Read from Cloud Storage

1. In the Cloud Data Fusion web interface, go to the Studiopage.
2. Click arrow_drop_down Sourceand select Cloud Storage. The node for a Cloud Storage source appears in the pipeline.

3. On the Cloud Storage node, click Properties.

   

4. In the Reference namefield, enter a name.

5. In the Pathfield, enter ${input.path} . This macro controls what the Cloud Storage input path will be in the different pipeline runs.

6. In the right Output Schema panel, remove the offsetfield from the output schema by clicking the trash icon in the offset field row.

   

7. Click Validateand address any errors.

8. Click the close to exit the Propertiesdialog.

Transform your data

1. In the Cloud Data Fusion web interface, go to your data pipeline on the Studiopage.
2. In the Transformdrop-down menu arrow_drop_down , select Wrangler.
3. In the Pipeline Studio canvas, drag an arrow from the Cloud Storage node to the Wrangler node.

   

4. Go to the Wrangler node in your pipeline and click the Properties.
5. In the Input field name, enter body .
6. In the Recipefield, enter ${directives} . This macro controls what the transform logic will be in the different pipeline runs.

   

7. Click Validateand address any errors.
8. Click the close to exit the Propertiesdialog.

Write to Cloud Storage

1. In the Cloud Data Fusion web interface, go to your data pipeline on the Studiopage.
2. In the Sinkdrop-down menu arrow_drop_down , select Cloud Storage.
3. In the Pipeline Studio canvas, drag an arrow from the Wrangler node to the Cloud Storage node that you just added.

   

4. Go to the Cloud Storage sink node in your pipeline and click Properties.
5. In the Reference namefield, enter a name.
6. In the Pathfield, enter the path of a Cloud Storage bucket in your project, where your pipeline can write the output files. If you don't have a Cloud Storage bucket, create one .

   

7. Click Validateand address any errors.
8. Click the close to exit the Propertiesdialog.

Set the macro arguments

1. In the Cloud Data Fusion web interface, go to your data pipeline on the Studiopage.
2. In the arrow_drop_down Conditions and Actionsdrop-down menu, click GCS Argument Setter.
3. In the Pipeline Studio canvas, drag an arrow from the Cloud Storage Argument Setter node to the Cloud Storage source node.

   

4. Go to the Cloud Storage Argument Setter node in your pipeline and click Properties.

5. In the URLfield, enter the following URL:

    gs://reusable-pipeline-tutorial/args.json 
   

   The URL corresponds to a publicly accessible object in Cloud Storage that contains the following content:

    {
     "arguments" : [
       {
         "name": "input.path",
         "value": "gs://reusable-pipeline-tutorial/user-emails.txt"
       },
       {
         "name": "directives",
         "value": "send-to-error !dq:isEmail(body)"
       }
     ]
   } 
   

   The first of the two arguments is the value for input.path . The path gs://reusable-pipeline-tutorial/user-emails.txt is a publicly accessible object in Cloud Storage that contains the following test data:

     alice 
    @example 
    . 
    com 
    bob 
    @example 
    . 
    com 
    craig 
    @invalid@example 
    . 
    com 
    
   

   The second argument is the value for directives . The value send-to-error !dq:isEmail(body) configures Wrangler to filter out any lines that are not a valid email address. For example, craig@invalid@example.com is filtered out.

6. Click Validateto make sure you don't have any errors.

7. Click the close to exit the Propertiesdialog.

Deploy and run your pipeline

1. In the top bar of the Pipeline Studiopage, click Name your pipeline. Name your pipeline and click Save.

2. Click Deploy.

3. To open the Runtime Argumentsand view the macro (runtime) input.path and directives arguments, click the arrow_drop_down drop-down next to Run.

   Leave the value fields blank to notify Cloud Data Fusion that the Cloud Storage Argument Setter node in the pipeline will set the values of these arguments during runtime.

   

4. Click Run.