Introduction In today’s fast-paced digital landscape, the demand for real-time analytics has shifted from a luxury to a foundational requirement. Traditional batch processing, with its inherent latency and overnight data loads, is no longer sufficient to power the agility required by the modern enterprise. As organizations strive to become truly data-driven, data engineering teams face a daunting challenge: how to seamlessly integrate highly structured, heterogeneous data from multiple operational systems such as ERPs, CRMs, and BPM applications into a unified Data Lakehouse. Historically, achieving this meant building and maintaining a tangled, fragile web of individual, table-specific pipelines. This legacy approach quickly becomes a maintenance nightmare, resulting in exorbitant operational overhead and bottlenecks whenever source schemas evolve or new tables are added. To overcome this integration bottleneck, modern Data Engineering demands a consolidated, intelligent approach to streaming data. Enter the "holy trinity" of real-time data architecture: Apache Kafka for resilient, high-throughput ingestion, Apache Flink for stateful, complex event processing, and Apache Iceberg serving as the robust, open table format for the Data Lake. By harnessing these core components within a unified ecosystem like the Cloudera Data Platform, engineering teams can design a single, dynamic pipeline capable of reading multiplexed data streams from various operational systems. Instead of sprawling infrastructure, this architecture utilizes a unified Flink application to perform inline validations and intelligently route records to their respective Iceberg tables on the fly. The result is a clean, scalable enterprise Data Lakehouse that eliminates pipeline sprawl while ensuring mission-critical use cases such as fraud detection, and compliance systems are fueled by high-quality, up-to-the-second data. The Architectural Challenge The Multi-Table Dilemma: Escaping Pipeline Sprawl In a traditional data ingestion setup, moving data from an operational system to a data warehouse often follows a rigid, one-to-one mapping strategy. Data engineering teams historically build distinct data pipelines for each of the data classes/tables within upstream ERP/CRM systems. While this approach is manageable for a handful of datasets, enterprise systems contain hundreds, often thousands of underlying tables. Scaling the traditional approach means deploying and monitoring a separate Kafka topic, a distinct stream processing job, and an isolated destination table for every single source entity. This leads to severe "pipeline sprawl." The operational overhead becomes paralyzing; infrastructure costs skyrocket due to underutilized, dedicated compute resources, and the CI/CD lifecycle grinds to a halt. Every time an upstream application adds a new table or modifies a schema, engineers must manually design, test, and deploy a brand-new pipeline. To achieve true real-time agility, enterprises must move away from this fragile paradigm and adopt a dynamic, multiplexed architecture where a single, unified pipeline can process and route hundreds of different tables simultaneously. The Need for Resilience When dealing with high-velocity streaming data from diverse operational systems, data quality is never guaranteed. Upstream systems frequently undergo unannounced schema evolutions, send unexpected null values, or transmit malformed payloads. In a naive streaming architecture, these anomalies are catastrophic. A single schema mismatch or typecasting error can throw an unhandled exception that crashes the entire stream processing job. Alternatively, the consumer might get stuck in an infinite retry loop, halting the flow of all subsequent data. When your business relies on real-time data for fraud detection, compliance audits, or live operational dashboards, a pipeline crash is a critical incident. The entire downstream ecosystem is starved of data while engineers scramble to parse application logs, isolate the offending record, and manually restart the pipeline. Therefore, a fault-tolerant approach is non-negotiable for enterprise data. Modern pipelines must be engineered with defensive processing capabilities. They need the resilience to seamlessly catch bad data, quarantine it for later inspection, and allow the continuous flow of healthy records without missing a beat. Architecture Walkthrough: The Pipeline in Action High-Level Overview: Looking at the enterprise Data Lakehouse architecture diagram below, you can see a departure from the traditional, rigid "one pipeline per table" model. Instead, we have constructed a unified, multiplexed streaming flow. This architecture consolidates data movement into a single, high-throughput thoroughfare, leveraging the native integration capabilities of Kafka, Flink, and Iceberg. The result is an elegant, end-to-end pipeline that handles ingestion, robust processing, and intelligent delivery across the entire enterprise data footprint. Let's break down exactly how a record travels through this system.     The Ingestion Tier - A Unified Funnel for Enterprise Data The journey begins at the Data Source Tier, where operational systems like your ERP, CRM, and BPM applications reside. Instead of deploying separate extraction processes for every individual table (e.g., customers, orders, invoices), change data capture (CDC) tools or event publishers stream these varied records directly into a single, unified Inbound Kafka Topic. By multiplexing different datasets into a common ingestion point, Kafka acts as a massive, decoupled shock absorber. It easily handles the fluctuating, high-velocity throughput of enterprise transactions without overwhelming downstream systems. Each message entering this topic carries not just the raw data payload, but crucial metadata, most importantly, the identifier of its source system, table, operation type(insert/update/delete) etc. Stateful Processing & Stream Splitting with Apache Flink Once the raw, mixed data lands in the Kafka topic, the Apache Flink Streaming Application takes over as the processing engine. Flink consumes the continuous stream and acts as a strict, real-time gatekeeper, performing inline data and schema validations on every single record. Because Flink supports stateful processing, it maintains complex validation logic in-memory. As a record flows through, the pipeline executes a series of checks (e.g., schema validation, type checking, and data integrity). Flink’s Side Outputs feature But what happens when a record inevitably fails these checks? This is where Flink’s Side Outputs feature becomes the hero of the architecture. Instead of throwing an exception and crashing the pipeline or silently dropping the data, Flink uses side outputs to elegantly split the pipeline into two distinct data streams: a main stream for validated "good" records, and a branched stream for "bad" records. Handling the "Bad" Records via Side Outputs Records that fail validation are seamlessly shunted into the side output stream, ensuring they never contaminate the downstream analytics tables. To maintain complete auditability and enable future reprocessing, this "bad records" data stream is simultaneously written to two destinations: An Error Iceberg Table: The malformed records are written to a centralized error_records_table. This allows data engineers to use standard SQL to query, investigate, and identify patterns in the bad data (e.g., catching an unannounced schema change from an upstream ERP).  The records are also published to an "Error Kafka Topic," which can trigger real-time alerts or feed into automated data correction and backfill workflows. The "Dynamic" Iceberg Sink Intelligent Routing for "Good" Records Focusing back on the main data stream of successfully validated, "good" records, we reach the final delivery phase. Traditionally, you would need to hardcode a specific sink for every table destination. Instead, this pipeline employs a Dynamic Iceberg Table Sink. As Flink processes the good records stream, it inspects each payload to extract the table_name metadata field. Using this extracted identifier, Flink dynamically routes the record on the fly to its corresponding Apache Iceberg destination. Behind the scenes, the Hive Metastore (HMS) plays a critical role as the central catalog. Flink communicates with HMS to seamlessly resolve the destination table's schema, metadata, and physical storage location (whether that happens to be HDFS on-premises or cloud storage like S3, ADLS, or GCS). The dynamic sink then intelligently writes the data. This means that if your source ERP system adds fifty new tables tomorrow, you don't need to write fifty new Flink sinks; as long as the tables are registered in the Hive Metastore, the dynamic routing logic will handle them automatically. Key Design Considerations & Best Practices Checkpointing & Exactly-Once Semantics The Flink-Iceberg Commit Cycle: Flink relies on its checkpointing mechanism to trigger Iceberg commits. The combination of Flink's distributed snapshots and Iceberg's atomic commits guarantees end-to-end exactly-once semantics. Balancing the Checkpoint Interval: Tuning the checkpoint frequency is critical. Too short: Leads to pipeline overhead, backpressure, and the "small file problem" in Iceberg (excessive metadata and tiny data files). Too long: Increases end-to-end data latency (downstream users wait longer to see data) and inflates recovery times in the event of a failure. Best Practice: Aim for a checkpoint interval between 1 to 5 minutes for streaming Data Lakes, depending on your latency SLAs, and ensure adequate idle time between checkpoints to let the system breathe. Kafka Offset Management & Fault Tolerance Externalized Offset Configuration: Hardcoding Kafka starting offsets is an anti-pattern. Externalizing offset initializers (e.g., earliest, latest, specific timestamp, or committed group offsets) via configuration files or deployment variables is crucial. Disaster Recovery: This flexibility is what allows data engineering teams to seamlessly execute replay or backfill operations to recover from data loss, logic errors, or downstream corruption without rewriting code. Iceberg Write Strategies: MOR vs COW Copy-On-Write (COW): Rewrites entire data files when updates or deletes occur. It yields fast reads but suffers from slow, heavy write penalties. Merge-On-Read (MOR): Writes inserts, updates, and deletes as smaller delta files, which the read engine merges on the fly. This provides lightning-fast writes at the expense of heavier read overhead. Streaming Default: For a high-throughput Flink streaming pipeline capturing CDC (Change Data Capture) or frequent updates, Merge-On-Read (MOR) is the definitive choice to prevent write bottlenecks. Additional Operational Best Practices (The "Missing Pieces") To ensure this architecture remains performant over time, consider adding these practices to your design: Automated Iceberg Maintenance: Because streaming with MOR generates thousands of delta files and snapshots, automated maintenance is mandatory. You must schedule asynchronous batch jobs (e.g., via Spark or Flink batch) to perform: Data Compaction (RewriteDataFiles): Merging small delta files into larger, read-optimized files. Snapshot Expiration (ExpireSnapshots): Removing old table states to keep metastore operations fast and reduce storage costs. Orphan File Deletion: Cleaning up unreferenced files from failed commits. Caution: There is a risk of losing data during orphan file deletion, so it has to be carried out with caution & a dry-run to list out candidates before deleting. Handling Schema Evolution: Flink and Iceberg handle schema evolution well, but your pipeline needs a strategy for when upstream ERPs drop columns or change data types. Utilize Iceberg's native schema evolution features, but ensure your Flink validation tier catches incompatible breaking changes before they hit the sink. Monitoring & Observability: Actively monitor Kafka Consumer Lag (to ensure Flink is keeping up) and Flink Checkpoint Duration/Size (to catch state bloat). If checkpoint times start creeping up, it is a leading indicator of pipeline distress. Storage Flexibility One of the greatest strengths of Apache Iceberg is its ability to completely abstract the physical storage layer from the data processing engine. Because Iceberg relies on a file-backed metadata design rather than directory-based tracking, it doesn't care where the bytes actually live. This abstraction grants enterprises immense architectural freedom and a seamless path to hybrid or multi-cloud topology. The exact same Flink pipeline and Hive Metastore configuration can write seamlessly to on-premises infrastructure like the Hadoop Distributed File System (HDFS) or to modern cloud object stores, including Amazon S3, Azure Data Lake Storage (ADLS Gen2), and Google Cloud Storage (GCS). This ensures your real-time lakehouse remains future-proof, allowing you to migrate, replicate, or split data across environments without ever rewriting your streaming application logic. Use Case Scenarios Any ERP, CRM or BPM systems that have complex data models that require data warehouses populated in real-time for analytics. These use cases exists in all industries Insurance: telematics & claims CDC into policy/exposure Iceberg tables; PII controls via Ranger; late-event handling with event-time windows. Retail/e-commerce: clickstream + orders/fulfillment analytics in near-real time; catalog/price changes fanned out to dedicated tables. Energy/Utilities: AMI/SCADA ingestion, anomaly detection, and SLA dashboards backed by fresh Iceberg snapshots. Deployment on Cloudera (public cloud & on-prem) CDP Public Cloud Kafka via Cloudera Streams Messaging; Flink via Cloudera Streams Processing; SDX centralizes Ranger/Atlas; Iceberg tables on S3/ABFS/GCS with HMS or REST catalog. Autoscaling, cloud-native security, object storage economics. CDP Private Cloud (on-prem) Same stack, storage via S3-compatible or Ozone/HDFS; HMS catalog. Unified governance and consistent APIs for hybrid portability. Programming API Choice:  Java API fully supports Data Stream API & Flink’s RowData objects, but Python does not yet support these features of Flink.   Closing thoughts The side-outputs + Dynamic Iceberg Sink pattern turns one Flink pipeline into a robust multi-table, real-time loader with first-class error handling ideal for lakehouse-native analytics on Cloudera. With Iceberg’s ACID guarantees, SDX governance, and the same code running on on-prem or cloud, teams can standardize on a single architecture for streaming ingestion, windowing, and analytics at scale. ... View more

Introduction This document describes the details of Knox SSO integration with a SAML identity provider (For the purpose of this article, the IDP is Ping Federate). Here, Ping Federate acts as the identity provider and it has the ability to connect to any authentication source such as AD, LDAP, etc.   The following diagram depicts the process at a high level:    The SAML authentication process has three components  Identity Provider: Identity provider has the flexibility to authenticate users against an authentication source, and sends a SAML response XML back to the service provider Service Provider: Service provider is an application that requires users to authenticate Authentication source:  A directory service such as AD, LDAP. The service provider uses the authentication token and does not need to know the authentication source. As part of this integration, both IDP (Ping Fed) and SP (Knox server) exchange information for the handshake to work successfully. The following information is shared to IDP by SP: Service Provider EntityID= https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&client_name=SAML2Client Assertion Consumer service URL (IDP sends SAML response to this endpoint) = https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&client_name=SAML2Client Knox hostnames and IP addresses so that ports can be opened on the firewalls etc  The following information is shared to SP by IDP: IDP metadata URL Knox configuration Configure knoxsso.xml to use pac4j provider which has the identity provider metadata that allows Knox to communicate with IDP. A sample knoxsso XML is available in the Appendix section below. Configure the Service XML that uses the SSOCookieProvider mechanism to route the initial service request to KnoxSSO. Request the IDP team for the metadata XML file. You may receive the metadata URL from the IDP team. This URL can be added in the Knox SSO XML under the property: saml.identityProviderMetadataPath​ Conversely, the IDP Metadata URL can be entered on the browser. Copy the XML content and paste it in a file such as /etc/knox/conf/idp-metadata.xml. This file path can be used as a value for the property saml.identityProviderMetadataPath. Ensure that the file is owned by Knox: $ chown knox:knox /etc/knox/conf/idp-metadata.xml​ The saml.serviceProviderMetadataPath property should be set to a file path on Knox server such as /etc/knox/conf/sp-metadata.xml. The property should be in knoxsso.xml file however, the file does not exist. Ensure that the service provider EntityID is shared to IDP team and it matches what we have in knoxsso.xml configuration. The Service Provider EntityID look like the following: https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&amp;client_name=SAML2Client​  In our scenario, the above URL was shared with the IDP team for them to the update SP metadata. However, we ran into an issue where the Entity IDs did not match between SP & IDP. The SAML trace shows that the “amp;” was removed. The solution for this is to remove “amp;” from the SP Entity ID on IDP: SP Entity ID on knox: https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&amp;client_name=SAML2Client​ SP Entity ID shared to Ping Federate:  https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&client_name=SAML2Client​ Create a Knox topology XML that uses the SSOCookieProvider and lists the services that should be placed behind the Knox SSO. The example file is present in the Appendix section. Ensure that the knoxsso topology content has been added in Ambari Knox configuration in the Advanced-Knoxsso-topology section and restart the Knox service. Ensure that both topology files (knoxsso, service topology xml) are owned by the Knox user:  $ chown knox:knox /etc/knox/conf/topologies/knoxsso.xml $ chown knox:knox /etc/knox/conf/topologies/clustername-sso.xml (this file is in Appendix)​ Issues Encountered: ISSUE 1 The attempt to access a Knox URL failed with below error while Knox was trying to communicate to Ping Federate server: javax.servlet.ServletException: org.pac4j.core.exception.TechnicalException: java.lang.UnsupportedOperationException: trusted certificate entries are not password-protected This error indicates that the keystore has a certificate that is of type “trustedCertEntry”.  Pac4j expects all certificates in Knox keystore to be of type “PrivateKeyEntry”. The certificate was converted from "trustedCertEntry" to "PrivateKeyEntry".  To resolve this issue, do the following to create a PrivateKeyEntry certificate:  Obtain the public key and private key for the certificate. Create a PKCS12 formatted keystore from the above keys: $ openssl pkcs12 -export -in cert -inkey key -out pkcert.p12 -name pkcert.alias​ Import the cert from PKCS12 formatted cert into the Knox java keystone: $ keytool -importkeystore -deststorepass knoxgwpasswd -destkeypass knoxgwkeypass -destkeystore gateway.jks -srckeystore pkcert.p12 -srcstoretype PKCS12 -srcstorepass somepasswd -alias pkcert.alias​ Delete any certificates of kind "trustedCertEntry" from knox gateway.jks: $ keytool -delete -noprompt -alias xxxxxx -keystore gateway.jks -storepass xxxxxx​ Once this is done, restart Knox and test again by entering a Knox URL for a Hadoop service (Ranger, Atlas, Yarn UI, Spark UI, Zeppelin UI, etc. ) ISSUE 2 Another error encountered during our test is that upon entering the URL for service UI (yarn UI), the browser redirects to the IDP login page, and upon entering credentials, it takes the user back to the IDP login page. This issue occurred because Knox certificate was regenerated or recreated and the Knox public key was not replaced in the configuration of services like Yarn, MapReduce, Ranger, Spark, Atlas, etc.   To resolve this issue, ensure that whenever Knox certificate or Keystore is modified, the public key should be extracted and added into Yarn and other service configurations. Once all the above configuration is in place, the Knox SSO service URL can be entered on the browser: https://knoxhost.domain:8443/gateway/topology/yarn/ The topology in above URL is not the KNOXSSO topology. It should be another topology that has SSOCookieProvider and has service URL mappings.   Once you enter the local AD credentials, IDP (Ping Fed) validates the user credentials and sends the SAML response back to Knox. Knox SSO redirects back to Yarn UI.  Appendix   The following are the Knox gateway logs during successful round trips form Knox to Ping federate:   Knoxsso.xml <topology> <gateway> <provider> <role>federation</role> <name>pac4j</name> <enabled>true</enabled> <param> <name>pac4j.callbackUrl</name> <value>https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso</value> </param> <param> <name>clientName</name> <value>SAML2Client</value> </param> <param> <name>saml.identityProviderMetadataPath</name> <value>/etc/knox/conf/idp-metadata.xml</value> </param> <param> <name>saml.serviceProviderMetadataPath</name> <value>/etc/knox/conf/sp-metadata.xml</value> </param> <param> <name>saml.serviceProviderEntityId</name> <value>https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso?pac4jCallback=true&amp;client_name=SAML2Client</value> </param> </provider> </gateway> <service> <role>KNOXSSO</role> <param> <name>knoxsso.cookie.secure.only</name> <value>true</value> </param> <param> <name>knoxsso.token.ttl</name> <value>3600000</value> </param> <param> <name>knoxsso.redirect.whitelist.regex</name> <value>.*$</value> </param> </service> </topology>   Knox Topology XML to expose the services(clustername-sso.xml):  <topology> <generated>true</generated> <gateway> <provider> <role>webappsec</role> <name>WebAppSec</name> <enabled>true</enabled> <param> <name>cors.enabled</name> <value>true</value> </param> <param><name>xframe.options.enabled</name><value>true</value></param> </provider> <provider> <role>federation</role> <name>SSOCookieProvider</name> <enabled>true</enabled> <param> <name>sso.authentication.provider.url</name> <value>https://knoxhost.domain:8443/gateway/knoxsso/api/v1/websso</value> </param> </provider> <provider> <role>identity-assertion</role> <name>Default</name> <enabled>true</enabled> </provider> </gateway> <service> <role>ATLAS-API</role> <url>https://knoxhost.domain:21443</url> <url>https://knoxhost.domain:21443</url> </service> <service> <role>ATLAS</role> <url>https://knoxhost.domain:21443</url> <url>https://knoxhost.domain:21443</url> </service> <service> <role>RANGER</role> <url>https://knoxhost.domain:6182</url> </service> <service> <role>RANGERUI</role> <url>https://knoxhost.domain:6182</url> </service> <service> <role>YARNUI</role> <url>https://knoxhost.domain:8090</url> <url>https://knoxhost.domain:8090</url> </service> </topology> ... View more

Introduction This article assumes that KnoxSSO for NiFi UI is enabled and is working as expected.   To configure Knox topology in order to access NiFi Rest API using Knox based NiFi URL, do the following:     Access NiFi UI through Knox SSO based URL such as the following: https://<knox.fqdn>:8443/gateway/<nifi-topology>/nifi-app/nifi/​ This helps the users to expose only the Knox host and port (not the NiFi hosts) and also authenticate users via SSO before they successfully log in to NiFi UI. Once the SSO is enabled on NiFi UI, NiFi URL for the UI or the Rest API will automatically redirect to the SSO page. Usually, the NiFi Rest API access involves obtaining Bearer token and using it in subsequent API calls. With Knox proxy-based NiFi endpoint, this Bearer token would not work as Knox does not recognize this token. On top of that, the URL always redirects to the SSO URL.    To establish access to Knox based NiFi URLs, both Knox and NiFi need to be configured to generate the Knox JWT token and honor the JWT token while accessing NiFi Rest API. Configuration The following is the process to enable NiFi API access to NiFi instances that are protected by KnoxSSO: Knox Create a KNOXTOKEN service in one of the Knox topologies to allow users to extract the Knox token. Any existing topology can be used for this purpose. In this example, knoxsso.xml is used to add KNOXTOKEN service. Add the following content at the end of the Advanced KnoxSSO topology in Ambari UI and restart Knox service: <service> <role>KNOXTOKEN</role> <param> <name>knox.token.ttl</name> <value>18000000</value> </param> <param> <name>knox.token.audiences</name> <value>hdftopology</value> </param> <param> <name>knox.token.target.url</name> <value>https://knox.fqdn:8443/gateway/hdftopology</value> </param> </service> ​ Create a new Knox topology XML file for NiFi API access via Knox token. This topology should use the JWTProvider for authentication which honors the Knox token. Place this file in the /etc/knox/conf/topologies/ folder and ensure it is owned by the "Knox" user. <topology> <gateway> <provider> <role>federation</role> <name>JWTProvider</name> <enabled>true</enabled> <param> <name>knox.token.audiences</name> <value>hdftopology</value> </param> </provider> <provider> <role>identity-assertion</role> <name>Default</name> <enabled>true</enabled> </provider> <provider> <role>authorization</role> <name>XASecurePDPKnox</name> <enabled>true</enabled> </provider> </gateway> <service> <role>NIFI</role> <url>https://nifi-1.domain:9091</url> <url>https://nifi-2.domain:9091</url> <url>https://nifi-3.domain:9091</url> <param> <name>useTwoWaySsl</name> <value>true</value> </param> </service> <service> <role>NIFI-API</role> <url>https://nifi-1.domain:9091</url> <url>https://nifi-2.domain:9091</url> <url>https://nifi-3.domain:9091</url> <param> <name>useTwoWaySsl</name> <value>true</value> </param> </service> </topology>​ Run the following command on Knox server: $ chown knox:knox /etc/knox/conf/topologies/hdftopology.xml​ The moment the topology file is added, browse the Knox logs to ensure that the topology is activated successfully. NiFi In Nifi configuration, add a new entry for nifi.web.proxy.context.path for NiFi to allow this URL path: “gateway/hdftopology/nifi-app”. ​ This path uses the new JWTProvider topology that we created for NiFi API access. After making the changes, restart NiFi. Validation Generate Knox token by using the following CURL command: $curl -ivku <username>:<password> https://knox.fqdn:8443/gateway/knoxsso/knoxtoken/api/v1/token​​ Extract the “access token” so that it can be used in subsequent NiFi API calls. This access token is nothing but the Knox JWT token. This token should be used as the Bearer token in NiFi API calls. An example NiFi API call looks like the following: [[email protected] topologies]# curl -ivk -H "Authorization: Bearer xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" https://knox.fqdn:8443/gateway/hdftopology/nifi-app/nifi-api/flow/cluster/summary * About to connect() to knox.fqdn port 8443 (#0) * Trying 10.xx.xxx.xx... * Connected to knox.fqdn (10.xx.xxx.xx) port 8443 (#0) ... ... * Connection #0 to host knox.fqdn left intact {"clusterSummary":{"connectedNodes":"3 / 3","connectedNodeCount":3,"totalNodeCount":3,"connectedToCluster":true,"clustered":true}   ... View more

This document seeks to explain the configuration steps required to access NiFi UI using Knox SSO and along with Knox proxy.   This eliminates the need to access NiFi UI using the regular port 9091 on the NiFi node     Configure Knox SSO for NiFi:     Create the CA signed Knox certificate:   Ensure that Knox certificate is signed with the same CA that NiFi certificates are signed. By default, knox creates a self-signed certificate which needs to be replaced by CA signed cert.   Go to knox server and do the following: Copy the CA signed certificate from the knox host as gateway.jks in the knox keystore path. $ cp /etc/security/certs/keystore.jks /var/lib/knox/data-3.1.0.0-78/security/keystores/gateway.jks $ chown knox:knox /var/lib/knox/data-3.1.0.0-78/security/keystores/gateway.jks   Knox certificate’s keystore password should match the knox master secret (let us say, it is Hadoop*123)   Now, let us change the new knox certificate’s password: $ keytool -storepasswd -keystore /var/lib/knox/data-3.1.0.0-78/security/keystores/gateway.jks Enter existing password: <changeit> New password: <Hadoop*123> Re-enter new password:   Change key's password for knox certificate:   $ keytool -keypasswd  -alias gateway-identity -keystore /var/lib/knox/data-3.1.0.0-78/security/keystores/gateway.jks   enter keystore password: <Hadoop*123> enter key's existing password: <changeit> enter key's new password: <Hadoop*123>   Restart Knox service using ambari UI   Deploy Knox Certificate on NiFi Nodes:   Run the following commands on all NiFi nodes: Ensure that the correct knox server is used in the below command.   $ openssl s_client -connect knox.example.com:8443</dev/null| openssl x509 -out /usr/hdf/current/nifi/conf/knox.pem    $ chown nifi:nifi /usr/hdf/current/nifi/conf/knox.pem      Add Knox hostname as a user in NiFi UI:   This step is very important, and needs to be done before modifying nifi configurations and restarting nifi   Login to NiFi UI as the admin user and add Knox hostname as a user and add the proxy request privilege to the Knox host user. Once knox DN is added as user, go to policies, select the policy “proxy user requests” and click on add user button. Select the knox host user you just added and click on ADD.     NiFi Configuration changes in Ambari UI:   Login to HDF Ambari UI make the following configuration changes.   Get the Knox server DN from knox certificate. For this: Login to knox server and run the command:   $ keytool -list -keystore /var/lib/knox/data-3.1.0.0-78/security/keystores/gateway.jks -v Enter the keystore password: <Hadoop*123>   Look for the word “gateway-identity” in the output and capture the DN of the “owner”. It will look something like this CN=knox.host.example.com,       Add the DN obtained from above step, and add it as a node identity in NiFi configuration as shown below. The last entry in the screen shot is the knox DN:   Advanced nifi-ambari-ssl-config:   nifi.security.needClientAuth = true   Add an entry in the “node identities” check box on ambari nifi configuration page, and add the full DN of the knox host from the knox certificate.     Advanced nifi-properties:   Make the following changes in ambari. Please note that nifi.security.user.login.identity.provider should be empty when you configure for knox SSO. Usually we see “ldap-provider” in this property. If you see it, make it empty.     In the following configuration, replace “mytopo” with the name of the “topology” name. the knox host name and port should be entered in “nifi.web.proxy.host”     Restart NiFi service on ambari    Knox proxy configuration: Configure Knox proxy in your knox topology file by adding below entries in the file. The topology file should be owned by “knox” user.   <topology>   <gateway>     <provider>       <role>federation</role>       <name>SSOCookieProvider</name>       <enabled>true</enabled>       <param>         <name>sso.authentication.provider.url</name>         <value>https://knox.host.example.com:8443/gateway/knoxsso/api/v1/websso</value>       </param>     </provider>     <provider>       <role>identity-assertion</role>       <name>Default</name>       <enabled>true</enabled>     </provider>   </gateway>   <service>     <role>NIFI</role>     <url>https://nifi1.example.com:9091</url>     <url>https://nifi2.example.com:9091</url>     <url>https://nifi3.example.com:9091</url>     <param>       <name>useTwoWaySsl</name>       <value>true</value>     </param>   </service> </topology>     Add the NiFi IP addresses and host name into your windows hosts file (if you are using private IPs)   Once it is done, enter the following url in the browser:   https://knox.host.example.com:8443/gateway/mytopo/nifi-app/nifi   It will take you to the knox SSO page, and then you can enter the user credentials and it will take you to the NiFi UI canvas. ... View more

This article seeks to walk you through the process developed in order to classify a given set of images into one of the x number of categories with the help of training datasets (of images) & a deep learning image recognition model "InceptionV3" & RanomForest classification algorithm. The technologies used are tensorflow & spark on hadoop platform. It inovles the following modules: Technology Stack: Spark 2.1.0, Python 3.5, HDP 2.6.1 Prepare the Training datasets: 1. Install the ImageMagick package using "yum install imagemagick" 2. Obtain sample images from the customer and also the label that needs to be associated with them 2. Extract the label (metadata) from each image using "ImageMagick" tool and place each image into different folders prepareTrainingData.sh This is a bash script that extracts the metadata from the image and puts them into different bins say category 1 through 5. These images were manually classified by analyzing them with naked eye. These images will be used for training our model. #!/bin/bash IMAGES="/home/arun/image-classification/Images" UNLABELED="./UnlabeledImages.txt" TRAININGDATA="/home/arun/image-classification/TrainingData" rm "$UNLABELED" for i in $(ls -1 "$IMAGES") do for j in $(ls -1 "$IMAGES"/"$i") do rating=`identify -verbose "$IMAGES"/"$i"/"$j" | grep xmp:Rating | cut -d':' -f3` rtng=`echo "$rating" | awk '{$1=$1};1'` case "$rtng" in 1) echo " this is cat 1" cp "$IMAGES"/"$i"/"$j" "$TRAININGDATA"/cat1/ ;; 2) echo "this is cat 2" cp "$IMAGES"/"$i"/"$j" "$TRAININGDATA"/cat2/ ;; 3) echo "this is cat 3" cp "$IMAGES"/"$i"/"$j" "$TRAININGDATA"/cat3/ ;; 4) echo "this is cat 4" cp "$IMAGES"/"$i"/"$j" "$TRAININGDATA"/cat4/ ;; 5) echo "thi is cat 5" cp "$IMAGES"/"$i"/"$j" "$TRAININGDATA"/cat5/ ;; *) echo "this is someting else" echo "$j" >> "$UNLABELED" ;; esac done done Classify Images: 1. Install the python packages numpy, keras, tensorflow, nose, pillow, h5py, py4j on all the gateway & worker nodes of the cluster. You can use either pip or anaconda for this. 2. Start a pyspark session and download a spark deep learning library from Databricks that runs on top of tensorflow and uses other python packages that we installed before. This spark DL library provides an interface to perform functions such as reading images into a spark dataframe, applying the InceptionV3 model and extract features from the images etc., 3. In the pyspark session, read the images into a dataframe and split the images into training and test dataframes. 4. Create a spark ml pipeline and add the stages 1) ImageFeaturizer 2) RandomForest Classifier 5. Execute the fit function and obtain a model 6. Predict using the model & also calculate the prediction accuracy ### Fire up a pyspark session export PYSPARK_PYTHON=/opt/anaconda3/bin/python3 export SPARK_HOME=/usr/hdp/current/spark2-client $SPARK_HOME/bin/pyspark --packages databricks:spark-deep-learning:0.1.0-spark2.1-s_2.11 --master yarn --executor-memory 3g --driver-memory 5g --conf spark.yarn.executor.memoryOverhead=5120 ### Add the spark deep-learning jars into the classpath import sys,glob,os sys.path.extend(glob.glob(os.path.join(os.path.expanduser("~"),".ivy2/jars/*.jar"))) ### PySpark code to read images, create spark ml pipeline, train the mode & predict from sparkdl import readImages from pyspark.sql.functions import lit from pyspark.ml.classification import RandomForestClassifier from pyspark.ml import Pipeline from sparkdl import DeepImageFeaturizer from pyspark.ml.evaluation import MulticlassClassificationEvaluator img_dir = "/user/arun/TrainingData" cat1_df = readImages(img_dir + "/cat1").withColumn("label", lit(1)) cat2_df = readImages(img_dir + "/cat2").withColumn("label", lit(2)) cat3_df = readImages(img_dir + "/cat3").withColumn("label", lit(3)) cat4_df = readImages(img_dir + "/cat4").withColumn("label", lit(4)) cat5_df = readImages(img_dir + "/cat5").withColumn("label", lit(5)) //Split the images where 90% of them go to training data, 10% go to test data cat1_train, cat1_test = cat1_df.randomSplit([0.9, 0.1]) cat2_train, cat2_test = cat2_df.randomSplit([0.9, 0.1]) cat3_train, cat3_test = cat3_df.randomSplit([0.9, 0.1]) cat4_train, cat4_test = cat4_df.randomSplit([0.9, 0.1]) cat5_train, cat5_test = cat5_df.randomSplit([0.9, 0.1]) train_df = cat1_train.unionAll(cat2_train).unionAll(cat3_train).unionAll(cat4_train).unionAll(cat5_train) test_df = cat1_test.unionAll(cat2_test).unionAll(cat3_test).unionAll(cat4_test).unionAll(cat5_test) featurizer = DeepImageFeaturizer(inputCol="image", outputCol="features", modelName="InceptionV3") rf = RandomForestClassifier(labelCol="label", featuresCol="features") p = Pipeline(stages=[featurizer, rf]) p_model = p.fit(train_df) predictions = p_model.transform(test_df) predictions.select("filePath", "label", "prediction").show(200,truncate=False) preds_vs_labels = predictions.select("prediction", "label") evaluator = MulticlassClassificationEvaluator(metricName="accuracy") print("accuracy of predictions by model = " + str(evaluator.evaluate(preds_vs_labels))) # TRY TO CLASSIFY CAT 5 IMAGES, AND SEE HOW CLOSE THEY GET IN PREDICTING cat5_imgs = readImages(img_dir + "/cat5").withColumn("label", lit(5)) pred5 = p_model.transform(cat5_imgs) pred5.select("filePath","label","prediction").show(200,truncate=False) Reference: https://medium.com/linagora-engineering/making-image-classification-simple-with-spark-deep-learning-f654a8b876b8 ... View more