Stream Catalog

The key to unlocking the value within data in motion and increasing productivity across an organization is a self-service tool for data discovery.

Available through both the Confluent Cloud UI and API, stream catalog allows users across teams to collaborate within a centralized, organized library designed for sharing, finding, and understanding the data needed to drive their projects forward quickly and efficiently. It’s like a digital library for data in motion allowing any user, experienced with Kafka or not, to search for what they need, find what’s already been built, and put it to use right away.

With classifications, teams can constantly increase the value of the business’s stream catalog by adding contextual details to data; for example, by labeling schema fields as “PII” (personal identifiable information) or “Sensitive”.

Searching Data and Schemas

Stream catalog in Confluent Cloud centralizes all schemas-related metadata and makes it available for discovery via the global search bar. In particular, with regard to schemas you can search for:

  • Schema subject name
  • Schema record name
  • Schema field name

Search with Filters

Click See all.. or hit return on your initial search results to view a more detailed list, with an option apply more filters.

This view shows “All Results” for a search on stocks.

../_images/dg-global-search-list-all.png

You can filter this further by selecting an entity on the left menu; for example, “Schemas”.

../_images/dg-global-search-schemas-filtered-list.png

Also, you can use the filters across the top to filter by “Environment”, “Tag”, or “Entities”.

This example narrows the search result based on entity type to show only schema subject names that include stocks.

../_images/dg-catalog-filter-by-schema-subject.png

This example filters on schema subjects, records, and fields in the demos environment that are tagged with my_stocks.

../_images/dg-catalog-filter-multiple.png

You can change the search on this detailed list view to search for other names of a selected entity type. For example, if you worked through the steps in the Quick Start to create a schema for the employees topic, search for employees in schemas. (Make sure you clear any other filters like tags that may not match the employees schema.)

../_images/dg-global-search-schemas-filtered-list-new.png

Tagging Data and Schemas

Stream catalog features enables entity tagging. Tags are searchable like any other entity.

A fundamental aspect of governance is the ability to organize data based on a shared vocabulary, including multiple concepts and categories. Confluent Cloud now provides the option to create and apply tags to schemas and fine-grained entities like data records and fields. On this version of Confluent Cloud, you can:

  • Create instances of provided tags (Public, Private, Sensitive, PII) and custom (“free form”) tags
  • Associate tags with schema versions, records, and fields
  • Apply multiple tags to a single field, record, or schema version

Create tags

To create a tag:

  1. Navigate to the Schema Registry tab on the settings for a selected Environment.

    ../_images/dg-tags-view-manage.png
  2. Click View & manage tags.

  3. Click Create tag, select either free-form or commonly used tags.

    • If you chose commonly used tags, select one or more tags that you want to create. (PII, which is personally identifiable information, Sensitive, Private, Public).
    • If you chose free-form, provide a tag name and optional description. Tag names must start with an alphabetic character, and can then include alpha characters, numbers and underscores.
    ../_images/dg-tags-name-rules.png

    When the name and description is properly filled in, the Create button is active. Click Create to create the tag with the current name and description.

    ../_images/dg-tags-create.png
  4. View available tags.

    After a tag is created, it shows in the list under Tag management.

    ../_images/dg-tags-management-list.png

Apply tags

To apply a tag:

  1. Navigate to a schema, data record, or data field for which you want to apply a tag.

    There are a few of different ways to do this:

    • From the Search bar, start typing the name of a schema, record, field, or topic where you want to apply a tag.
    • From the same environment level view on the Schema Registry tab, click View and manage schemas and select a schema from the list.
    • From within a cluster, navigate to a topic, then click the Schema tab for that topic.
  2. On the Schemas tab for a topic, you can add tags as follows.

    • To add tags to selected schema version as a whole, select Add tags to this version on the right panel, and select a tag from the drop-down list of available tags.
    ../_images/dg-tags-schema-version.png
    • To add tags to a selected record or field, expand the tree view of the schema (the default view), click the plus icon image_reference next to the entity, and select a tag from the drop-down list of available tags.
    ../_images/dg-tags-data-field.png
  3. View applied tags.

    Applied tags show next to the schema version, records, and fields with which they are associated.

Remove a tag from a schema version or data field

To remove a previously applied tag from an entity:

  1. Navigate to the schema that includes the tag.
  2. Click the x on the applied tag.

Edit a tag

To edit a tag description:

  1. Navigate to the Schema Registry tab on the settings for a selected Environment.
  2. Click View & manage tags.
  3. Select the tag you want to edit from the Tag management list.
  4. Click the edit icon next to the description, edit, and click Save.

Tip

You cannot edit the name of an existing tag, only its description. To rename a tag, remove it from any entities to which it is applied, delete the tag, and create a new one.

Delete a tag

If you want to delete a tag, first make sure that the tag is not currently applied to any entities. If the tag is in use, the delete operation will fail.

To delete a tag from an environment:

  1. Navigate to the Schema Registry tab on the settings for a selected Environment.
  2. Click View & manage tags.
  3. Select the tag you want to delete from the Tag management list.
  4. Click the trashcan icon in the upper right.
    • If the tag is in use (applied to one or more entities), you will get a warning and the tag will not be deleted.
    • If the tag is not in use, it is deleted.

Catalog API Documentation

The stream catalog API allows you to create, retrieve, update, and delete catalog entities. To learn more, see the Confluent Cloud CATALOG API (V1).

Note

The catalog API provides search capabilities for schema entity types (subjects, records, and fields) only. The API currently does not include search capabilities for other entity types like topics, connectors, and so forth; whereas, the global search in the Confluent Cloud UI searches all entity types.

Catalog API Usage Examples

This section provides examples of calls to the Confluent Cloud CATALOG API (V1) using curl commands.

For your API testing, you may want to use jq along with --silent flag for curl to get nicely formatted output for the given commands. Some of the initial examples below show commands with these options included, and the expected output. If you don’t use --silent and jq, the content of the output will be the same, just unformatted.

Setup

You’ll need the following to make API calls to the Catalog API:

  • API endpoint URL for the Schema Registry cluster in the environment you want to use. In the examples below, this is referred to as the <SCHEMA-REGISTRY-URL>. You can find this on Confluent Cloud UI by navigating to the Schema Registry tab for your environment, or log on the Confluent Cloud command line, by navigating to your environment and entering ccloud ccloud schema-registry cluster describe. (A handy list of CLI commands is here.)
  • API key and secret for Schema Registry in the environment you want to use. (If needed, see Create an API Key for Confluent Cloud Schema Registry.) Schema subjects live at the level of an environment, in per-environment Schema Registry clusters, so just choose the environment that contains the Kafka clusters you want to use. In the examples below, these are referred to as <API-KEY>:<API-SECRET>; you will provide these as colon-separated pair in the curl commands.

Create tag definition

curl -u <API-KEY>:<API-SECRET> \
--header 'Content-Type: application/json' \
--data '[ { "entityTypes" : [ "sr_schema", "sr_record", "sr_field" ],"name" : "<TAG_NAME>","description" : "Test API call"} ]' \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs

For example, to create one of the commonly-used, pre-fab tags (PII, Sensitive, Private, Public), provide that tag name in the API call. The following command creates a PII tag:

curl -u <API-KEY>:<API-SECRET> \
--header 'Content-Type: application/json' \
--data '[ { "entityTypes" : [ "sr_schema", "sr_record", "sr_field" ],"name" : "PII","description" : "Personally Identifiable Information"} ]' \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs

To create a custom tag called PCI with which to tag stock symbols on your buy list, using curl --silent and piping it through jq:

curl --silent -u <API-KEY>:<API-SECRET> \
--header 'Content-Type: application/json' \
--data '[ { "entityTypes" : [ "sr_schema", "sr_record", "sr_field" ],"name" : "PCI","description" : "Payment Card Industry Data Security Standard"} ]' \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs | jq

Your output should resemble:

"category": "CLASSIFICATION",
  "createdBy": "root",
  "updatedBy": "root",
  "createTime": 1631914851910,
  "updateTime": 1631914851910,
  "version": 1,
  "name": "PCI",
  "description": "Payment Card Industry Data Security Standard",
  "typeVersion": "1.0",
  "attributeDefs": [],
  "superTypes": [],
  "entityTypes": [
    "sr_field",
    "sr_schema",
    "sr_record"
  ]

To create a custom tag called stocks_buy with which to tag stock symbols on your buy list, using curl --silent and piping it through jq:

curl --silent -u <API-KEY>:<API-SECRET> \
--header 'Content-Type: application/json' \
--data '[ { "entityTypes" : [ "sr_schema", "sr_record", "sr_field" ],"name" : "stocks_buy","description" : "stock symbols on buy list"} ]' \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs | jq

Your output should resemble:

"category": "CLASSIFICATION",
  "createdBy": "root",
  "updatedBy": "root",
  "createTime": 1631914483922,
  "updateTime": 1631914483922,
  "version": 1,
  "name": "stocks_buy",
  "description": "stock symbols on buy list",
  "typeVersion": "1.0",
  "attributeDefs": [],
  "superTypes": [],
  "entityTypes": [
    "sr_field",
    "sr_schema",
    "sr_record"
  ]

Also, you can check the Confluent Cloud UI to verify that the new tags show up there.

Update a tag definition by adding attributes (key, value pairs)

curl -u <API-KEY>:<API-SECRET> --request PUT \
  --url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs \
  --header 'Content-Type: application/json' \
  --data '[{ "entityTypes" : [ "sr_record", "sr_field" ],
           "name" : "PCI", "description" : "Payment Card Industry Data Security Standard" ,
           "attributeDefs" : [ { "name" : "key", "cardinality" : "SINGLE", "typeName" : "string" },
           { "name" : "region", "isOptional" : "true", "cardinality" : "SINGLE", "typeName" : "string" },
           { "name" : "keytype", "isOptional" : "true", "cardinality" : "SINGLE", "typeName" : "string" } ] }]'

Your output should resemble:

"category": "CLASSIFICATION",
 "name": "PCI",
 "description": "Payment Card Industry Data Security Standard",
 "attributeDefs": [
   {
     "name": "key",
     "typeName": "string",
     "isOptional": false,
     "cardinality": "SINGLE",
     "valuesMinCount": -1,
     "valuesMaxCount": -1,
     "isUnique": false,
     "isIndexable": false,
     "includeInNotification": false,
     "searchWeight": -1
   },
   {
     "name": "region",
     "typeName": "string",
     "isOptional": true,
     "cardinality": "SINGLE",
     "valuesMinCount": -1,
     "valuesMaxCount": -1,
     "isUnique": false,
     "isIndexable": false,
     "includeInNotification": false,
     "searchWeight": -1
   },
   {
     "name": "keytype",
     "typeName": "string",
     "isOptional": true,
     "cardinality": "SINGLE",
     "valuesMinCount": -1,
     "valuesMaxCount": -1,
     "isUnique": false,
     "isIndexable": false,
     "includeInNotification": false,
     "searchWeight": -1
   }
 ],
 "superTypes": [],
 "entityTypes": [
   "sr_field",
   "sr_record"
 ]

List tag definitions

curl -u <API-KEY>:<API-SECRET> \
  --request GET \
  --url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs

Your output will resemble:

"category": "CLASSIFICATION",
    "guid": "f94803c3-6aba-4b41-a003-2f8fc1957c29",
    "createdBy": "root",
    "updatedBy": "root",
    "createTime": 1631593629096,
    "updateTime": 1631593629096,
    "version": 1,
    "name": "PII",
    "description": "Personally identifiable information",
    "typeVersion": "1.0",
    "attributeDefs": [],
    "superTypes": [],
    "entityTypes": [
      "cf_entity"
    ]
  },
  {
    "category": "CLASSIFICATION",
    "createdBy": "root",
    "updatedBy": "root",
    "createTime": 1631914851910,
    "updateTime": 1631914851910,
    "version": 1,
    "name": "PCI",
    "description": "Payment Card Industry Data Security Standard",
    "typeVersion": "1.0",
    "attributeDefs": [],
    "superTypes": [],
    "entityTypes": [
      "sr_field",
      "sr_schema",
      "sr_record"
    ]
  },

  {
    "category": "CLASSIFICATION",
    "createdBy": "root",
    "updatedBy": "root",
    "createTime": 1631914483920,
    "updateTime": 1631914483920,
    "version": 1,
    "name": "stocks_buy",
    "description": "stock symbols on buy list",
    "typeVersion": "1.0",
    "attributeDefs": [],
    "superTypes": [],
    "entityTypes": [
      "sr_field",
      "sr_schema",
      "sr_record"
    ]
  },

  ...

Get tag definition

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/types/tagdefs/PII

Your output should resemble:

 "category": "CLASSIFICATION",
"createdBy": "root",
"updatedBy": "root",
"createTime": 1631593629105,
"updateTime": 1631593629105,
"version": 1,
"name": "PII",
"description": "Personally identifiable information",
"typeVersion": "1.0",
"attributeDefs": [],
"superTypes": [],
"entityTypes": [
  "cf_entity"
]

Search fields by name

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_field&query=credit_card

Search fields by tag

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url '<SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_field&tag=PII'

Search fields by name and tag

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url '<SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_field&query=credit_card&tag=PII' \

Search schema by name

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url '<SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_schema&query=com.mycorp.mynamespace.SampleRecord' \

Search schema by tag

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url '<SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_schema&tag=PII'

Search record by name

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url '<SCHEMA-REGISTRY-URL>/catalog/v1/search/basic?type=sr_record&query=SampleRecord'

Tag field in Avro

curl -u <API-KEY>:<API-SECRET> \
--request POST \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/entity/tags \
--header 'Authorization: Basic ...' \
--header 'Content-Type: application/json' \
--data '[ {  "entityType" : "sr_field",
  "entityName" : "lsrc-0xvxr9:.:100049:com.mycorp.mynamespace.fooRecord:my_field1",
  "typeName" : "PII"} ]'

Tag field in Avro with attributes

curl -u <API-KEY>:<API-SECRET> \
--request POST \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/entity/tags \
--header 'Content-Type: application/json' \
--data '[ {    "entityType" : "sr_field",
  "entityName" : "lsrc-0xvxr9:.:100049:com.mycorp.mynamespace.fooRecord:my_field1",
  "typeName" : "PCI",
      "attributes" : { "key" : "XPTO3343333" } } ]'

Get the tag attributes from a field

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/entity/type/sr_field/name/100049:com.mycorp.mynamespace.fooRecord:my_field1/tags

Tag schema version

curl -u <API-KEY>:<API-SECRET> \
--request POST \
--url <SCHEMA-REGISTRY-URL>/catalog/v1/entity/tags \
--header 'Content-Type: application/json' \
--data '[ {    "entityType" : "sr_schema",  "entityName" : "lsrc-0xvxr9:.:100002",  "typeName" : "PII"  } ]'

Get a schema given a subject name

curl -u <API-KEY>:<API-SECRET> \
--request GET \
--url 'https://<SCHEMA-REGISTRY-URL>/catalog/v1/search/attribute?type=sr_subject_version&attrName=name&attrValuePrefix=big'