Canalización de índice de catálogo de Ingest

La siguiente secuencia de pasos ilustra la canalización de indexación de Catálogo implementada en Apache NiFi. El flujo consta principalmente de dos etapas:

1. Crear un documento de catálogo
2. Asociar filtros de Catálogo

Esta etapa describe cómo transformar los datos del catálogo y cargarlos en el índice del catálogo. Empieza con la ejecución del SQL siguiente para recuperar datos de Catálogo de la base de datos de Commerce.

SELECT C.CATALOG_ID, C.IDENTIFIER, C.MEMBER_ID, D.NAME, D.SHORTDESCRIPTION, 
            COALESCE(D.LANGUAGE_ID, L.LANGUAGE_ID) LANGUAGE_ID, 
            L.LOCALENAME, T.MASTERCATALOG, F.STOREENT_ID
		  FROM LANGUAGE L, STORELANG F, STORECAT T, CATALOG C
          LEFT OUTER JOIN CATALOGDSC D ON(C.CATALOG_ID = D.CATALOG_ID AND D.LANGUAGE_ID = ${param.langId})
		 WHERE T.STOREENT_ID IN (SELECT RELATEDSTORE_ID FROM STOREREL WHERE STATE = 1 AND STRELTYP_ID = -4 AND STORE_ID = ${param.storeId})
            AND T.CATALOG_ID = C.CATALOG_ID  AND C.CATALOG_ID = ${param.catalogId}
		    AND F.STOREENT_ID = ${param.storeId} AND F.LANGUAGE_ID = L.LANGUAGE_ID AND L.LANGUAGE_ID = ${param.langId}	

Para ver un ejemplo de código, consulte Ejemplos de la etapa 1.

SELECT F.STOREENT_ID, F.CATALOG_ID, F.USAGE, F.CATFILTER_ID, F.IDENTIFIER, X.EXPRESSION_ID, X.MEMBER_ID,
	       CAST(X.QUERY AS VARCHAR) QUERY, X.TRADING_ID
		  FROM EXPRESSION X, CATFILTER F
	WHERE X.CATFILTER_ID = F.CATFILTER_ID
	AND F.CATALOG_ID IN (SELECT C.CATALOG_ID FROM STORECAT C​ WHERE C.MASTERCATALOG = 1
		                           AND C.STOREENT_ID IN (SELECT RELATEDSTORE_ID FROM STOREREL
			                                              WHERE STATE = 1 AND STRELTYP_ID = -4 AND STORE_ID = ${param.storeId}))
        AND F.STOREENT_ID IN ( SELECT RELATEDSTORE_ID FROM STOREREL WHERE STATE = 1 AND STRELTYP_ID = -32 AND STORE_ID = ${param.storeId} )
​    

A continuación, el conjunto de resultados se pasa al procesador FindFiltersFromDatabase​ para su transformación, utilizando la tabla siguiente para ​correlacionar el campo de base de datos devuelto desde el SQL anterior con un campo de índice en el índice de Catálogo:​Para ver un ejemplo de código, consulte Ejemplos de la etapa 2.

El código siguiente es un ejemplo de los datos de entrada para el procesador CreateCatalogDocumentFromDatabase:

{
  "CATALOG_ID": 10001,
  "IDENTIFIER": "Extended Sites Catalog Asset Store",
  "MEMBER_ID": 7000000000000001000,
  "NAME": "Extended Sites Catalog Asset Store",
  "SHORTDESCRIPTION": null,
  "LANGUAGE_ID": -1,
  "LOCALENAME": "en_US           ",
  "MASTERCATALOG": "1",
  "STOREENT_ID": 1
}

El procesador CreateCatalogDocumentFromDatabase transforma los datos de entrada en los siguientes datos de salida:

{ "update": { "_id": "1--1-10001", "_index": ".auth.1.catalog.202006160325", 
"retry_on_conflict": 5, "_source": false } }
{
  "doc": {
    "identifier": {
      "catalog": "Extended Sites Catalog Asset Store",
      "specification": "catalog",
      "language": "en_US"
    },
    "name": {
      "normalized": "Extended Sites Catalog Asset Store",
      "raw": "Extended Sites Catalog Asset Store"
    },
    "id": {
      "catalog": "10001",
      "member": "7000000000000001000",
      "language": "-1",
      "store": "1"
    },
    "type": "master",
    "__meta": {
      "created": "2020-07-28T18:42:27.926Z",
      "modified": "2020-07-28T18:42:27.942Z",
      "version": {
        "min": 0,
        "max": 0
      }
    }
  },
  "doc_as_upsert": true
}

El código siguiente es un ejemplo de los datos de entrada para el procesador FindFiltersFromDatabase.

{
  "STOREENT_ID": 1,
  "CATALOG_ID": 10001,
  "USAGE": null,
  "CATFILTER_ID": 3074457345616679000,
  "IDENTIFIER": "TestCatalogFilter",
  "EXPRESSION_ID": 3074457345618259500,
  "MEMBER_ID": null,
  "QUERY": "( +*:* -parentCatgroup_id_search:\"10001_10006\")",
  "TRADING_ID": null
}

{ "update": { "_id": "1--1-10001", "_index": ".auth.1.catalog.202006160325", 
"retry_on_conflict": 5, "_source": false } }
{
  "doc": {
    "filters": [
      {
        "identifier": "TestCatalogFilter",
        "usage": "",
        "query": "( +*:* -path.tree:\"10006\")",
        "id": {
          "filter": "3074457345616679000",
          "expression": "3074457345618259500",
          "contract": "",
          "member": ""
        }
      }
    ],
    "__meta": {
      "modified": "2020-07-28T18:54:46.440Z"
    }
  }
}