Copy a catalog using the Data Processing Library

Objectives: Copy the GeoJSON contents of a source catalog using the Data Processing Library.

Complexity: Intermediate

Time to complete: 30 min

Prerequisites: Verify Your Credentials; Verify Maven Settings, Organize your work in projects

Source code: Download

This example demonstrates how to create a catalog using the OLP CLI, configure it with a new versioned layer, populate the layer with GeoJSON contents, and write a Data Processing Library application to copy these contents to another catalog.

Create the source catalog

The OLP CLI requires a valid set of HERE credentials to authenticate itself with the Data Services, so make sure that the Verify your credentials tutorial returns the expected result.

If you have not already done so, download and unzip the Java and Scala examples and the CLI. Add the tools/OLP_CLI folder of the unzipped file to your ${PATH}.

Give your source catalog a unique identifier and name, for example, {{YOUR_USERNAME}}-geojson-dpl-tutorial.

Replace {{YOUR_SRC_CATALOG_ID}} that follows with your own identifier and {{YOUR_PROJECT_HRN}} with Project HRN from Organize your work in projects tutorial, and run the following command:

olp catalog create {{YOUR_SRC_CATALOG_ID}} "{{YOUR_USERNAME}} geojson example" \
    --summary "geojson example" \
    --description "geojson example" \
    --scope {{YOUR_PROJECT_HRN}}

The CLI returns as follows:

Catalog {{YOUR_SRC_CATALOG_HRN}} has been created.

Add a layer to contain the raw GeoJSON contents. This versioned layer has content type "application/vnd.geo+json", "heretile" partitioning at zoom level 12, and covers the administrative area of "CN" (China).

olp catalog layer add {{YOUR_SRC_CATALOG_HRN}} geojson "geojson example" \
    --versioned \
    --content-type=application/vnd.geo+json \
    --partitioning=heretile:12 \
    --coverage=CN \
    --summary "geojson example" \
    --description "geojson example"
    --scope {{YOUR_PROJECT_HRN}}

The CLI returns as follows:

Layer GeoJSON has been added to the catalog.

Make a folder called source-partitions/cn.

mkdir source-partitions/cn -p

Make a file in source-partitions/cn named 24355949 and populate it with the following GeoJSON. This represents a polygon inside partition 24355949.

{
   "type": "FeatureCollection",
   "features": [{
       "type": "Feature",
       "geometry": {
           "type": "Polygon",
           "coordinates": [[
           [116.33837761622146, 39.982829219791704],
           [116.35978234479171, 39.96142449122147],
           [116.35978234479171, 39.93115363377853],
           [116.33837761622146, 39.909748905208296],
           [116.30810675877854, 39.909748905208296],
           [116.28670203020829, 39.93115363377853],
           [116.28670203020829, 39.96142449122147],
           [116.30810675877854, 39.982829219791704],
           [116.33837761622146, 39.982829219791704]
           ]]
       },
       "properties": {
           "tooltip": "GREEN",
           "style" : {
                "color" : "#228B22",
                "fill" : "#228B22",
                "opacity": 1.0
            },
            "width": 5
       }
   }]
}

Make a file in source-partitions/cn named 24355960 and populate it with the following GeoJSON. This represents a polygon inside partition 24355960.

{
   "type": "FeatureCollection",
   "features": [{
       "type": "Feature",
       "geometry": {
           "type": "Polygon",
           "coordinates": [[
           [116.42626824122146, 39.982829219791704],
           [116.44767296979171, 39.96142449122147],
           [116.44767296979171, 39.93115363377853],
           [116.42626824122146, 39.909748905208296],
           [116.39599738377854, 39.909748905208296],
           [116.37459265520829, 39.93115363377853],
           [116.37459265520829, 39.96142449122147],
           [116.39599738377854, 39.982829219791704],
           [116.42626824122146, 39.982829219791704]
           ]]
       },
       "properties": {
           "tooltip": "GREEN",
           "style" : {
                "color" : "#228B22",
                "fill" : "#228B22",
                "opacity": 1.0
            },
            "width": 5
       }
   }]
}

Upload your GeoJSON partitions to the GeoJSON layer of your catalog:

olp catalog layer partition put {{YOUR_SRC_CATALOG_HRN}} geojson --input=./source-partitions/cn --scope {{YOUR_PROJECT_HRN}}

The CLI returns as follows:

Partitions 24355960, 24355949 were successfully uploaded.

You can inspect the content of the partitions in the platform portal. To do this, navigate to the portal, find {{YOUR_SRC_CATALOG_HRN}} in the list, and open this catalog. Select the geojson layer to view its details and inspect its data in the Inspect tab.

You can see outlines for your two partitions in the Beijing area on the map.

To render a green octagon, zoom in and click on these partitions. To open a tooltip with the text "GREEN," hold your mouse over the octagon.

Also, you can list partitions with the OLP CLI by executing the following command:

olp catalog layer partition list {{YOUR_SRC_CATALOG_HRN}} geojson --scope {{YOUR_PROJECT_HRN}}

For more details on working with raw GeoJSON content as Workspace Data, see this document.

Create the destination catalog

Give your destination catalog a unique identifier and name, for example, {{YOUR_USERNAME}}-geojson-copy-dpl-tutorial.

Replace {{YOUR_DST_CATALOG_ID}} as shown in the following example with your own identifier, and create the destination catalog and geojson layer in the same way as the source catalog shown previously.

olp catalog create {{YOUR_DST_CATALOG_ID}} "{{YOUR_USERNAME}} geojson example copy"
    --summary "geojson example copy"
    --description "geojson example copy"
    --scope {{YOUR_PROJECT_HRN}}
olp catalog layer add {{YOUR_DST_CATALOG_HRN}} geojson "geojson example" \
    --versioned \
    --content-type=application/vnd.geo+json \
    --partitioning=heretile:12 \
    --coverage=CN \
    --summary "geojson example copy"
    --description "geojson example copy"
    --scope {{YOUR_PROJECT_HRN}}

Alternatively, you can query the layer configuration of the GeoJSON layer of your source catalog as json, and then use that to configure your destination catalog.

olp catalog layer show {{YOUR_SRC_CATALOG_HRN}} geojson --json --scope {{YOUR_PROJECT_HRN}}

The CLI returns as follows:

{
    "coverage": {"adminAreas": ["CN"]},
    "summary": "geojson example",
    "volume": {"volumeType": "Durable"},
    "layerType": "Versioned",
    "billingTags": [],
    "name": "geojson example",
    "contentEncoding": "",
    "description": "geojson example",
    "partitioning": {
        "scheme": "heretile",
        "tileLevels": [12]
    },
    "id": "geojson",
    "contentType": "application/vnd.geo+json",
    "tags": []
}

You can then add the output into a config.json file to update your catalog with the GeoJSON layer configuration.:

{
  "id": {{YOUR_CATALOG_ID}},
  "name": "{{YOUR_USERNAME}} geojson example copy",
  "summary": "geojson example copy",
  "description": "geojson example copy",
  "layers": [
    "// json output from the "olp catalog layer show" command above"
  ]
}

olp catalog update {{YOUR_DST_CATALOG_HRN}} --config config.json --scope {{YOUR_PROJECT_HRN}}

Add a layer to contain the internal compilation state. The Data Processing Library uses this layer for stateful processing in subsequent runs. For more details on the state layer, see the Developer's Guide.

olp catalog layer add {{YOUR_DST_CATALOG_HRN}} state "internal compilation state" \
    --content-type="application/octet-stream" \
    --versioned --partitioning=generic \
    --summary "internal compilation state" \
    --description "internal compilation state" \
    --scope {{YOUR_PROJECT_HRN}}

Copy the GeoJSON layer contents

Create the following folder structure for the project:

copy-geojson
└── src
    └── main
        ├── java
        └── resources
        └── scala

To do this with a single bash command:

mkdir -p copy-geojson/src/main/{java,resources,scala}

The POM for this example is identical to that of the first Maven example, except for its parent and dependencies section:

<parent>
    <groupId>com.here.platform</groupId>
    <artifactId>sdk-batch-bom_2.12</artifactId>
    <version>2.54.3</version>
    <relativePath/>
</parent>

<dependencies>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>batch-core_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>pipeline-runner_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>batch-catalog-dataservice_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.pipeline</groupId>
        <artifactId>context-logging_${scala.compat.version}</artifactId>
    </dependency>
</dependencies>

<dependencies>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>batch-core-java_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>pipeline-runner-java_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.data.processing</groupId>
        <artifactId>batch-catalog-dataservice_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_${scala.compat.version}</artifactId>
    </dependency>
    <dependency>
        <groupId>com.here.platform.pipeline</groupId>
        <artifactId>context-logging_${scala.compat.version}</artifactId>
    </dependency>
</dependencies>

In the resources folder, create a file called application.conf and fill it with these contents:

here.platform.data-processing.driver {
  appName = "CopyCatalogScalaTutorial"
}

Configure the logger using the following contents for resources/log4j.properties:

# Root logger configuration
log4j.rootLogger=INFO, console

# Direct log messages to console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yyyy/MM/dd HH:mm:ss.SSS} %-7p %c{1}: %m%n

# Hide Spark informative logging
log4j.logger.org.apache.spark=WARN
log4j.logger.org.spark_project=WARN

The code for a 1:1 compiler to copy the GeoJSON contents is as follows:

/*
 * Copyright (c) 2018-2023 HERE Europe B.V.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.here.platform.data.processing.blobstore.{Payload, Retriever}
import com.here.platform.data.processing.build.BuildInfo
import com.here.platform.data.processing.catalog.Catalog
import com.here.platform.data.processing.catalog.Layer
import com.here.platform.data.processing.compiler._
import com.here.platform.data.processing.driver.config.CompleteConfig
import com.here.platform.data.processing.driver.runner.pipeline.PipelineRunner
import com.here.platform.data.processing.driver.{
  DriverBuilder,
  DriverContext,
  DriverSetupWithBuilder
}
import com.here.platform.pipeline.logging.{ContextLogging, LogContext}
import com.here.platform.data.processing.spark.partitioner.Partitioner

object CopyCatalogScala extends PipelineRunner with DriverSetupWithBuilder {

  type Data = Array[Byte]

  // This identifier must match that of the command-line parameter for the input catalog, i.e.,
  // -Dpipeline.config.input-catalogs.src.... and -Dpipeline.job.catalog-versions.input-catalogs.src...
  val srcCatalogId = Catalog.Id("src")

  val geojsonLayerId = Layer.Id("geojson")

  class Compiler(retriever: Retriever)
      extends Direct1ToNCompiler[Data]
      with InputLayers
      with OutputLayers
      with ContextLogging {

    // The compiler only processes a single input geojson layer.
    override def inLayers: Map[Catalog.Id, Set[Layer.Id]] = Map(srcCatalogId -> Set(geojsonLayerId))

    // The compiler outputs to a single output geojson layer.
    override def outLayers: Set[Layer.Id] = Set(geojsonLayerId)

    // Directly maps a partition from the input catalog to the partition with the same layer ID and partition name in the output catalog.
    override def mappingFn(inKey: InKey): Iterable[OutKey] =
      Iterable(inKey.copy(catalog = outCatalogId))

    // Extracts the raw geojson input for a single partition, as a byte array. This will then be passed
    // to compileOutFn with the corresponding output key from mappingFn.
    override def compileInFn(in: (InKey, InMeta)): Data = {
      logger.info("compileInFn (" + in.key + ")")
      retriever.getPayload(in.key, in.meta).content
    }

    // Writes out the extracted geojson byte array as a payload for the given output key.
    override def compileOutFn(outKey: OutKey, intermediate: Data): Option[Payload] = {
      logger.info("compileOutFn (" + outKey + ")")
      Some(Payload(intermediate))
    }

    // No custom partitioning needed
    override def inPartitioner(parallelism: Int): Option[Partitioner[InKey]] = None
    override def outPartitioner(parallelism: Int): Option[Partitioner[OutKey]] = None
  }

  val applicationVersion: String = BuildInfo.version

  def configureCompiler(completeConfig: CompleteConfig,
                        context: DriverContext,
                        builder: DriverBuilder): builder.type = {

    val taskBuilder = builder.newTaskBuilder("copycatalogscala")
    val compiler = new Compiler(context.inRetriever(srcCatalogId))
    builder.addTask(taskBuilder.withDirect1ToNCompiler(compiler).build())
  }
}

/*
 * Copyright (c) 2018-2023 HERE Europe B.V.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.here.platform.data.processing.build.BuildInfo;
import com.here.platform.data.processing.driver.runner.pipeline.java.PipelineRunner;
import com.here.platform.data.processing.java.Pair;
import com.here.platform.data.processing.java.blobstore.Payload;
import com.here.platform.data.processing.java.blobstore.Retriever;
import com.here.platform.data.processing.java.catalog.partition.Key;
import com.here.platform.data.processing.java.catalog.partition.Meta;
import com.here.platform.data.processing.java.compiler.Direct1ToNCompiler;
import com.here.platform.data.processing.java.compiler.InputLayers;
import com.here.platform.data.processing.java.compiler.OutputLayers;
import com.here.platform.data.processing.java.driver.Default;
import com.here.platform.data.processing.java.driver.DriverBuilder;
import com.here.platform.data.processing.java.driver.DriverContext;
import com.here.platform.data.processing.java.driver.TaskBuilder;
import com.here.platform.data.processing.java.driver.config.CompleteConfig;
import com.here.platform.data.processing.java.spark.partitioner.PartitionerOfKey;
import com.here.platform.pipeline.logging.java.ContextAwareLogger;
import java.util.*;

class Compiler implements Direct1ToNCompiler<byte[]>, InputLayers, OutputLayers {

  // This identifier must match that of the command-line parameter for the input catalog, i.e.,
  // -Dpipeline.config.input-catalogs.src.... and
  // -Dpipeline.job.catalog-versions.input-catalogs.src...
  private String srcCatalogId = "src";

  private String geojsonLayerId = "geojson";

  private Retriever retriever;

  private ContextAwareLogger contextAwareLogger;

  Compiler(DriverContext driverContext) {
    this.retriever = driverContext.inRetriever(srcCatalogId);
    this.contextAwareLogger = new ContextAwareLogger(getClass());
  }

  // The compiler only processes a single input geojson layer.
  @Override
  public Map<String, Set<String>> inLayers() {
    return Collections.unmodifiableMap(
        new HashMap<String, Set<String>>() {
          {
            put(
                srcCatalogId,
                new HashSet<String>() {
                  {
                    add(geojsonLayerId);
                  }
                });
          }
        });
  }

  // The compiler outputs to a single output geojson layer.
  @Override
  public Set<String> outLayers() {
    return Collections.unmodifiableSet(
        new HashSet<String>() {
          {
            add(geojsonLayerId);
          }
        });
  }

  // Directly maps a partition from the input catalog to the partition with the same layer ID and
  // partition name in the output catalog.
  @Override
  public Iterable<Key> mappingFn(Key inKey) {
    return Collections.singletonList(
        new Key(Default.OutCatalogId(), inKey.layer(), inKey.partition()));
  }

  // Extracts the raw geojson input for a single partition, as a byte array. This will then be
  // passed
  // to compileOutFn with the corresponding output key from mappingFn.
  @Override
  public byte[] compileInFn(Pair<Key, Meta> in) {
    contextAwareLogger.info("compileInFn (" + in.getKey() + ")");
    return retriever.getPayload(in.getKey(), in.getValue()).content();
  }

  // Writes out the extracted geojson byte array as a payload for the given output key.
  @Override
  public Optional<Payload> compileOutFn(Key outKey, byte[] intermediate) {
    contextAwareLogger.info("compileOutFn (" + outKey + ")");
    return Optional.of(new Payload(intermediate));
  }

  // No custom partitioning needed
  @Override
  public Optional<PartitionerOfKey> inPartitioner(int parallelism) {
    return Optional.empty();
  }

  // No custom partitioning needed
  @Override
  public Optional<PartitionerOfKey> outPartitioner(int parallelism) {
    return Optional.empty();
  }
}

public class CopyCatalogJava {
  public static void main(String... args) {
    new PipelineRunner() {

      @Override
      public String applicationVersion() {
        return BuildInfo.version();
      }

      @Override
      public DriverBuilder configureCompiler(
          CompleteConfig completeConfig, DriverContext context, DriverBuilder builder) {
        TaskBuilder taskBuilder = builder.newTaskBuilder("copycatalogjava");
        Compiler compiler = new Compiler(context);
        return builder.addTask(taskBuilder.withDirect1ToNCompiler(compiler, byte[].class).build());
      }
    }.main(args);
  }
}

Run the copy as a full compilation

To run the application locally, execute the following command:

mvn compile exec:java \
    -Dexec.mainClass=CopyCatalogScala \
    -Dpipeline.config.input-catalogs.src.hrn="{{YOUR_SRC_CATALOG_HRN}}" \
    -Dpipeline.config.output-catalog.hrn="{{YOUR_DST_CATALOG_HRN}}" \
    -Dpipeline.job.catalog-versions.input-catalogs.src.version=0 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.processing-type="reprocess" \
    -Dexec.args="--master local[*]" \
    -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}

mvn compile exec:java \
    -Dexec.mainClass=CopyCatalogJava \
    -Dpipeline.config.input-catalogs.src.hrn="{{YOUR_SRC_CATALOG_HRN}}" \
    -Dpipeline.config.output-catalog.hrn="{{YOUR_DST_CATALOG_HRN}}" \
    -Dpipeline.job.catalog-versions.input-catalogs.src.version=0 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.processing-type="reprocess" \
    -Dexec.args="--master local[*]" \
    -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}

After the local job is completed, you can inspect the copied contents in the platform portal. To do this, navigate to the portal, find {{YOUR_DST_CATALOG_HRN}} in the list and open this catalog. Select the geojson layer to view its details and inspect its data in the Inspect tab.

Update the source catalog

Make a new GeoJSON partition:

Make a folder called source-partitions/cn.

mkdir new-partitions/cn -p

Make a file in new-partitions/cn named 24355962 and populate it with the following GeoJSON.

{
   "type": "FeatureCollection",
   "features": [{
       "type": "Feature",
       "geometry": {
           "type": "Polygon",
           "coordinates": [[
           [116.42626824122146, 40.070719844791704],
           [116.44767296979171, 40.04931511622147],
           [116.44767296979171, 40.01904425877853],
           [116.42626824122146, 39.997639530208296],
           [116.39599738377854, 39.997639530208296],
           [116.37459265520829, 40.01904425877853],
           [116.37459265520829, 40.04931511622147],
           [116.39599738377854, 40.070719844791704],
           [116.42626824122146, 40.070719844791704]
           ]]
       },
       "properties": {
           "tooltip": "GREEN",
           "style" : {
                "color" : "#228B22",
                "fill" : "#228B22",
                "opacity": 1.0
            },
            "width": 5
       }
   }]
}

Upload your new partition to the GeoJSON layer of your source catalog:

olp catalog layer partition put {{YOUR_SRC_CATALOG_HRN}} geojson --input=./new-partitions/cn --scope {{YOUR_PROJECT_HRN}}

The CLI returns as follows:

Partition 24355962 was successfully uploaded.

After the local job is completed, you can inspect the copied contents in the platform portal. To do this, navigate to the portal, find {{YOUR_SRC_CATALOG_HRN}} in the list and open this catalog. Select the geojson layer to view its details and inspect its data in the Inspect tab.

Run incremental compilation

To run the application locally, execute the following command:

mvn compile exec:java \
    -Dexec.mainClass=CopyCatalogScala \
    -Dpipeline.config.input-catalogs.src.hrn="{{YOUR_SRC_CATALOG_HRN}}" \
    -Dpipeline.config.output-catalog.hrn="{{YOUR_DST_CATALOG_HRN}}" \
    -Dpipeline.job.catalog-versions.input-catalogs.src.since-version=0 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.version=1 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.processing-type="changes" \
    -Dexec.args="--master local[*]" \
    -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}

mvn compile exec:java \
    -Dexec.mainClass=CopyCatalogJava \
    -Dpipeline.config.input-catalogs.src.hrn="{{YOUR_SRC_CATALOG_HRN}}" \
    -Dpipeline.config.output-catalog.hrn="{{YOUR_DST_CATALOG_HRN}}" \
    -Dpipeline.job.catalog-versions.input-catalogs.src.since-version=0 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.version=1 \
    -Dpipeline.job.catalog-versions.input-catalogs.src.processing-type="changes" \
    -Dexec.args="--master local[*]" \
    -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}

Within the logging output, you can see lines like the ones below. Notice that only the new partition is encountered during compileInFn and compileOutFn.

2018/07/09 17:17:44.015 INFO    Direct1ToNCompilerExecutor: Incremental compilation execution // Task=copycatalogscala
2018/07/09 17:17:45.290 INFO    CopyCatalogScala$Compiler: compileInFn (Key('src,'geojson,24355962))
2018/07/09 17:17:45.655 INFO    CopyCatalogScala$Compiler: compileOutFn (Key('output,'geojson,24355962))

2018/07/09 17:27:09.383 INFO    Direct1ToNCompilerExecutor: Incremental compilation execution // Task=copycatalogjava
2018/07/09 17:27:11.070 INFO    Compiler: compileInFn (Key('src,'geojson,24355962))
2018/07/09 17:27:11.913 INFO    Compiler: compileOutFn (Key('output,'geojson,24355962))

Upon completion of the local job, you can inspect the copied contents on the portal at https://platform.hereolp.cn/data//geojson/inspect (replace YOUR_DST_CATALOG_HRN with your real catalog HRN).

Further information

For additional details on the topics covered in this tutorial, you can refer to the following source: Data Processing Library Developer Guide.