Inspect storage with sampling

The following examples demonstrate using the Cloud DLP API to scan a 90% subset of a Cloud Storage bucket for person names. The scan starts from a random location in the dataset and only includes text files under 200 bytes.

Explore further

For detailed documentation that includes this code sample, see the following:

Code sample


To learn how to install and use the client library for Cloud DLP, see Cloud DLP client libraries.

To authenticate to Cloud DLP, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

using Google.Api.Gax.ResourceNames;
using Google.Cloud.Dlp.V2;
using Google.Cloud.PubSub.V1;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class InspectStorageWithSampling
    public static async Task<DlpJob> InspectAsync(
        string projectId,
        string gcsUri,
        string topicId,
        string subId,
        Likelihood minLikelihood = Likelihood.Possible,
        IEnumerable<InfoType> infoTypes = null)

        // Instantiate the dlp client.
        var dlp = DlpServiceClient.Create();

        // Construct Storage config by specifying the GCS file to be inspected
        // and sample method.
        var storageConfig = new StorageConfig
            CloudStorageOptions = new CloudStorageOptions
                FileSet = new CloudStorageOptions.Types.FileSet
                    Url = gcsUri
                BytesLimitPerFile = 200,
                FileTypes = { new FileType[] { FileType.Csv } },
                FilesLimitPercent = 90,
                SampleMethod = CloudStorageOptions.Types.SampleMethod.RandomStart

        // Construct the Inspect Config and specify the type of info the inspection
        // will look for.
        var inspectConfig = new InspectConfig
            InfoTypes =
                infoTypes ?? new InfoType[] { new InfoType { Name = "PERSON_NAME" } }
            IncludeQuote = true,
            MinLikelihood = minLikelihood

        // Construct the pubsub action.
        var actions = new Action[]
            new Action
                PubSub = new Action.Types.PublishToPubSub
                    Topic = $"projects/{projectId}/topics/{topicId}"

        // Construct the inspect job config using above created objects.
        var inspectJob = new InspectJobConfig
            StorageConfig = storageConfig,
            InspectConfig = inspectConfig,
            Actions = { actions }

        // Issue Create Dlp Job Request
        var request = new CreateDlpJobRequest
            InspectJob = inspectJob,
            ParentAsLocationName = new LocationName(projectId, "global"),

        // We keep the name of the job that we just created.
        var dlpJob = dlp.CreateDlpJob(request);
        var jobName = dlpJob.Name;

        // Listen to pub/sub for the job
        var subscriptionName = new SubscriptionName(projectId, subId);
        var subscriber = await SubscriberClient.CreateAsync(

        await subscriber.StartAsync((PubsubMessage message, CancellationToken cancel) =>
            if (message.Attributes["DlpJobName"] == jobName)
                return Task.FromResult(SubscriberClient.Reply.Ack);
                return Task.FromResult(SubscriberClient.Reply.Nack);

        // Get the latest state of the job from the service
        var resultJob = dlp.GetDlpJob(new GetDlpJobRequest
            DlpJobName = DlpJobName.Parse(jobName)

        // Parse the response and process results.
        System.Console.WriteLine($"Job status: {resultJob.State}");
        System.Console.WriteLine($"Job Name: {resultJob.Name}");

        var result = resultJob.InspectDetails.Result;
        foreach (var infoType in result.InfoTypeStats)
            System.Console.WriteLine($"Info Type: {infoType.InfoType.Name}");
            System.Console.WriteLine($"Count: {infoType.Count}");
        return resultJob;


To learn how to install and use the client library for Cloud DLP, see Cloud DLP client libraries.

To authenticate to Cloud DLP, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class InspectGcsFileWithSampling {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String gcsUri = "gs://" + "your-bucket-name" + "/path/to/your/file.txt";
    String topicId = "your-pubsub-topic-id";
    String subscriptionId = "your-pubsub-subscription-id";
    inspectGcsFileWithSampling(projectId, gcsUri, topicId, subscriptionId);

  // Inspects a file in a Google Cloud Storage Bucket.
  public static void inspectGcsFileWithSampling(
      String projectId, String gcsUri, String topicId, String subscriptionId)
      throws ExecutionException, InterruptedException, IOException {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources.
    try (DlpServiceClient dlp = DlpServiceClient.create()) {
      // Specify the GCS file to be inspected and sampling configuration
      CloudStorageOptions cloudStorageOptions =

      StorageConfig storageConfig =

      // Specify the type of info the inspection will look for.
      // See for complete list of info types
      InfoType infoType = InfoType.newBuilder().setName("PERSON_NAME").build();

      // Specify how the content should be inspected.
      InspectConfig inspectConfig =

      // Specify the action that is triggered when the job completes.
      String pubSubTopic = String.format("projects/%s/topics/%s", projectId, topicId);
      Action.PublishToPubSub publishToPubSub =
      Action action = Action.newBuilder().setPubSub(publishToPubSub).build();

      // Configure the long running job we want the service to perform.
      InspectJobConfig inspectJobConfig =

      // Create the request for the job configured above.
      CreateDlpJobRequest createDlpJobRequest =
              .setParent(LocationName.of(projectId, "global").toString())

      // Use the client to send the request.
      final DlpJob dlpJob = dlp.createDlpJob(createDlpJobRequest);
      System.out.println("Job created: " + dlpJob.getName());

      // Set up a Pub/Sub subscriber to listen on the job completion status
      final SettableApiFuture<Boolean> done = SettableApiFuture.create();

      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      MessageReceiver messageHandler =
          (PubsubMessage pubsubMessage, AckReplyConsumer ackReplyConsumer) -> {
            handleMessage(dlpJob, done, pubsubMessage, ackReplyConsumer);
      Subscriber subscriber = Subscriber.newBuilder(subscriptionName, messageHandler).build();

      // Wait for job completion semi-synchronously
      // For long jobs, consider using a truly asynchronous execution model such as Cloud Functions
      try {
        done.get(15, TimeUnit.MINUTES);
      } catch (TimeoutException e) {
        System.out.println("Job was not completed after 15 minutes.");
      } finally {

      // Get the latest state of the job from the service
      GetDlpJobRequest request = GetDlpJobRequest.newBuilder().setName(dlpJob.getName()).build();
      DlpJob completedJob = dlp.getDlpJob(request);

      // Parse the response and process results.
      System.out.println("Job status: " + completedJob.getState());
      System.out.println("Job name: " + dlpJob.getName());
      InspectDataSourceDetails.Result result = completedJob.getInspectDetails().getResult();
      System.out.println("Findings: ");
      for (InfoTypeStats infoTypeStat : result.getInfoTypeStatsList()) {
        System.out.print("\tInfo type: " + infoTypeStat.getInfoType().getName());
        System.out.println("\tCount: " + infoTypeStat.getCount());

  // handleMessage injects the job and settableFuture into the message reciever interface
  private static void handleMessage(
      DlpJob job,
      SettableApiFuture<Boolean> done,
      PubsubMessage pubsubMessage,
      AckReplyConsumer ackReplyConsumer) {
    String messageAttribute = pubsubMessage.getAttributesMap().get("DlpJobName");
    if (job.getName().equals(messageAttribute)) {
    } else {


To learn how to install and use the client library for Cloud DLP, see Cloud DLP client libraries.

To authenticate to Cloud DLP, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

import threading  # noqa: F811, E402, I100

import  # noqa: F811, E402
import  # noqa: F811, E402

def inspect_gcs_with_sampling(
    project: str,
    bucket: str,
    topic_id: str,
    subscription_id: str,
    info_types: List[str] = None,
    file_types: List[str] = None,
    min_likelihood: str = None,
    max_findings: int = None,
    timeout: int = 300,
) -> None:
    """Uses the Data Loss Prevention API to analyze files in GCS by
    limiting the amount of data to be scanned.
        project: The Google Cloud project id to use as a parent resource.
        bucket: The name of the GCS bucket containing the file, as a string.
        topic_id: The id of the Cloud Pub/Sub topic to which the API will
            broadcast job completion. The topic must already exist.
        subscription_id: The id of the Cloud Pub/Sub subscription to listen on
            while waiting for job completion. The subscription must already
            exist and be subscribed to the topic.
        info_types: A list of strings representing infoTypes to look for.
            A full list of info type categories can be fetched from the API.
        file_types: Type of files in gcs bucket where the inspection would happen.
        min_likelihood: A string representing the minimum likelihood threshold
            that constitutes a match. One of: 'LIKELIHOOD_UNSPECIFIED',
        max_findings: The maximum number of findings to report; 0 = no maximum.
        timeout: The number of seconds to wait for a response from the API.

    # Instantiate a client.
    dlp =

    # Prepare info_types by converting the list of strings into a list of
    # dictionaries.
    if not info_types:
        info_types = ["FIRST_NAME", "LAST_NAME", "EMAIL_ADDRESS"]
    info_types = [{"name": info_type} for info_type in info_types]

    # Specify how the content should be inspected. Keys which are None may
    # optionally be omitted entirely.
    inspect_config = {
        "info_types": info_types,
        "exclude_info_types": True,
        "include_quote": True,
        "min_likelihood": min_likelihood,
        "limits": {"max_findings_per_request": max_findings},

    # Setting default file types as CSV files
    if not file_types:
        file_types = ['CSV']

    # Construct a cloud_storage_options dictionary with the bucket's URL.
    url = "gs://{}/*".format(bucket)
    storage_config = {
        "cloud_storage_options": {
            "file_set": {"url": url},
            "bytes_limit_per_file": 200,
            "file_types": file_types,
            "files_limit_percent": 90,
            "sample_method": 'RANDOM_START',

    # Tell the API where to send a notification when the job is complete.
    topic =, topic_id)
    actions = [{"pub_sub": {"topic": topic}}]

    # Construct the inspect_job, which defines the entire inspect content task.
    inspect_job = {
        "inspect_config": inspect_config,
        "storage_config": storage_config,
        "actions": actions,

    # Convert the project id into full resource ids.
    parent = f"projects/{project}/locations/global"

    # Call the API
    operation = dlp.create_dlp_job(
        request={"parent": parent, "inspect_job": inspect_job}
    print("Inspection operation started: {}".format(

    # Create a Pub/Sub client and find the subscription. The subscription is
    # expected to already be listening to the topic.
    subscriber =
    subscription_path = subscriber.subscription_path(project, subscription_id)

    # Set up a callback to acknowledge a message. This closes around an event
    # so that it can signal that it is done and the main thread can continue.
    job_done = threading.Event()

    def callback(message):
            if message.attributes["DlpJobName"] ==
                # This is the message we're looking for, so acknowledge it.

                # Now that the job is done, fetch the results and print them.
                job = dlp.get_dlp_job(request={"name":})
                print(f"Job name: {}")
                if job.inspect_details.result.info_type_stats:
                    for finding in job.inspect_details.result.info_type_stats:
                            "Info type: {}; Count: {}".format(
                      , finding.count
                    print("No findings.")

                # Signal to the main thread that we can exit.
                # This is not the message we're looking for.
        except Exception as e:
            # Because this is executing in a thread, an exception won't be
            # noted unless we print it manually.

    # Register the callback and wait on the event.
    subscriber.subscribe(subscription_path, callback=callback)
    finished = job_done.wait(timeout=timeout)
    if not finished:
            "No event received before the timeout. Please verify that the "
            "subscription provided is subscribed to the topic provided."

What's next

To search and filter code samples for other Google Cloud products, see the Google Cloud sample browser.