Community Articles

How to Submit Spark Application through Livy REST APIApache Livy supports using REST APIs to submit Spark applications, it is quite similar to use “spark-submit” in vanilla Spark. In this article we will briefly introduce how to use Livy REST APIs to submit Spark applications, and how to transfer existing “spark-submit” command to REST APIs.Using spark-submitIn vanilla Spark, normally we should use “spark-submit” command to submit Spark application to a cluster, a “spark-submit” command is like:./bin/spark-submit \ --class <main-class> \ --master <master-url> \ --deploy-mode <deploy-mode> \ --conf <key>=<value> \ ... # other options <application-jar> \ [application-arguments]Some of the key options are: --class: The entry point for your application (e.g. org.apache.spark.examples.SparkPi).--master: The master URL for the cluster (e.g. spark://23.195.26.187:7077).--deploy-mode: Whether to deploy your driver on the worker nodes (cluster) or lo

File processing + autoscaling seems to have an antinomic relationship. It doesn't have to be..really.  Autoscaling often drives inefficient behaviors..why? The concomitant response... "It Autoscales". Autoscaling still requires sound design principles. Without that, It won't autoscale well.   Autoscaling within a distributed framework requires proper data dichotomization along with processing/service layer decoupling Anti-Pattern Taking an example of where I've seen the most amount of heartburn. Large files (i.e. zip/tar/etc) land in s3/Storage area. The knee-jerk reaction is to feed it into a distributed processing engine that autoscales. The "autoscaling" part appropriately. It may. Most likely you're flipping a coin and lighting a candle, hoping that all works out. What if the file sizes are heterogeneous and the variances between them could be quite significant? What about error handling? Does it have to be all or nothing (meaning all files need to be processi

Recently I was tasked to build a possible pattern regarding how to handle slowing changing dimensions (Type 2 to be specific) within CDP. The obvious answer is to use Hive ACID, but clearly, that answer is too generic. I needed to build a pipeline similar to what I used to do as an Informatica developer and verify legitimacy to the solution/pattern. Here we go. Why? The objective was to build a possible (one of many) pattern on how to handle SCD Type 2 dimensions with CDP.  Outcome: I was able to repeat a typical ETL workflow with CDP.  How? By taking off my horse blinders...Really. What are SCDs? For those who may not be family with slowly changing dimensions within a EDW context, here is a great quick read: Types Of Dimension Tables How? Data lands in a sourcing area Super typical Pull source data & run it through a cleaning/processing layer for staging Ready for merging against SCD table Run an ACID merge between stage and SCD Table Not rocket scien

This article outlines the steps needed to setup ODBC access to Hive via Apache Knox from a Linux workstation. While there seems to be some "reasonable" documentation other there on setting up this access from Windows, it took quite some time to figure out how to do this from Linux. For this example, I am installing the Hortonworks Linux ODBC driver on CentOS Linux release 7.2.1511 then verifying it works using both the isql ODBC command line tool and PyODBC. Installation Steps 1) Install the UnixODBC Dependencies sudo yum install unixODBC 2) Download and Install Hortonworks ODBC Linux Driver I developed this guide using the below version of the driver. Note that the following URL is for the CentOS 7 version of the driver. If you are running an older version of the OS, then visit https://hortonworks.com/downloads/ and use the CentOS 6 driver. Also note that the below URL will likely change as new versions of the driver are released so it is a good idea to check the Hortonworks downloa

In this article, we will learn how to register a Hive UDFs using Spark HiveWarehouseSession. Download and build the Spark Hive UDF example.git clone https://github.com/rangareddy/spark-hive-udfcd spark-hive-udfmvn clean package -DskipTests Copy the target/spark-hive-udf-1.0.0-SNAPSHOT.jar to the edge node. Login to edge node and upload the spark-hive-udf-1.0.0-SNAPSHOT.jar to HDFS location for example, /tmp.hdfs dfs -put ./brickhouse-0.7.1-SNAPSHOT.jar /tmp Launch the spark-shell with 'hwc' parameters.spark-shell \ --jars /opt/cloudera/parcels/CDH/jars/hive-warehouse-connector-assembly-*.jar \ --conf spark.sql.hive.hiveserver2.jdbc.url='jdbc:hive2://hiveserver2_host1:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2' \ --conf spark.sql.hive.hwc.execution.mode=spark \ --conf spark.datasource.hive.warehouse.metastoreUri='thrift://metastore_host:9083' \ --conf spark.datasource.hive.warehouse.load.staging.dir='/tmp' \ --conf spark.datasource.hive.warehouse.us

Money management and retirement planning are extremely important issues. The conventional wisdom says to get a Financial Advisor who will invest your money and allow you to reach your financial goals. However, as of late, financial institutions have began to offer services generally referred to as Digital Advisors or "Robo Advisors". These services aim to offer a very low cost, broad spectrum, long term investment advice across a limited set of financial products. The advice is personalized based on basic input from the customer such as level of risk tolerance, initial investment, years to invest, ect. The basic premise behind the offering is diversification. Statistics have shown that few money managers are able to beat the S&P 500 over the long term, especially after fees. The Digital Advisor aims to spread the initial investment across a broad spectrum of securities at a much lower cost. The Digital Advisor provides another major advantage over the Human Financial Advisor. Using

ContextAs of CDP 7.1.2, Sentry is deprecated with Kudu, and Ranger becomes the solution for Fine-Grained Authorization in Kudu. The question this article is meant to address is what does it mean for the Impala/Kudu stack to enable Ranger for Kudu, especially if Impala is already using Ranger? This is not obvious up front because the information is spread over three different sets of documents. Fundamentally the answer is nothing. The reason for this is straightforward: enabling the Kudu module in Ranger should automatically configure the --trusted_user_acl flag in Kudu to include Impala (if installed) and Hive (if installed). What this flag does is exempt the listed users from being checked against Kudu's authorization model, which in CDP installations mean Ranger policies. So, the correct way to enforce access to Impala and Hive tables, which are stored as Kudu, is the Hadoop-SQL policies set for Impala and Hive. By default, enabling Ranger for Kudu should have no impact on

Continuing my series of how-to articles for CDP, today we explore how to connect to Impala via JDBC in JSON. In my example, I will use a Jupyter notebook running in CML, but this can be generalized. This process is actually fairly easy, so let's dive in. Step 1: Setup Impala JDBC drivers First, download the latest impala JDBC drivers from Cloudera JDBC Driver 2.6.17 for Impala. Then, upload them to your machine. Here is an example of a CML Jupyter sessions with the jars uploaded: Finally, make sure that you set up your CLASSPATH properly by opening a terminal session and typing the following: CLASSPATH=.:/home/cdsw/ImpalaJDBC4.jar:/home/cdsw/ImpalaJDBC41.jar:/home/cdsw/ImpalaJDBC42.jarexport CLASSPATH​ Step 2: Install JayDeBeApi To install JayDeBeApi, run the following:pip3 install JayDeBeApi ​ A recommended step to avoid getting an error along the lines of "AttributeError: type object 'java.sql.Types' has no attribute '__javaclass__'", would be to downgrade your jpype by runn

The file "cdswCluster.runs.jsonlines" in the output contents of "cdsw logs" contains session information of all projects. But there should be a regular archiving operation, that is, it may only keep a fixed number of records/ records within a fixed time range. I can't confirm this now. I have sorted out how to query this information through the curl command: On the CDSW Master node:   authHeader="Service-Authorization: Basic $(printf cdsw-metrics-service:$SERVICE_ACCOUNT_SECRET | base64 -w 0)"domain=$(kubectl get secrets internal-secrets --namespace=${CDSW_NAMESPACE} -o jsonpath="{.data.domain}" | base64 -d)apiUrl="${protocol}://${domain}/api/v1"datasets_json=$(curl -sSf -H "$authHeader" "${apiUrl}/metrics/datasets/names")if [ 0 -eq "$?" ]; then datasets=$(echo "$datasets_json" | python2 -c "import sys, json; print '\n'.join(json.load(sys.stdin))") for dataset_name in $datasets; do curl -sSf -H "$authHeader" "${apiUrl}/metrics/datasets/values/${dataset_name}" >$(pwd

In past years many companies have come up with (Hybrid) Cloud strategies, and there is no shortage of recommendations for when and how to use Cloud providers such as AWS, Azure, GCP, and others. However, most advice is geared towards a generic setting, and may not translate well to a Big Data context.    And here is the main reason why:   BIG Data requires BIG resources that have a BIG cost impact   This does not mean that any specific vendor is expensive, and it is not limited to a single deployment form such as PaaS or SaaS. It comes down to the fundamental economics that infrastructure provides must cover the costs they make, and if you use their solution for a big amount of data, these costs will be larger than for a small lightweight app.    In this context, hopefully, the following advice makes sense.   Avoid Network UsageFlexible Cloud Infrastructure vs Economical On-Prem infrastructureCostValueStay in control of your dataSecurityAccessibi

In CDP Public Cloud, Data Hub is a service for launching and managing workload clusters. Data Hub clusters strongly resemble traditional (CDH/HDP) clusters. This makes Data Hub clusters an attractive destination for migrating existing workloads to the cloud. In CDP Data Hub, clusters are secure by default.  Among other things, this means that perimeter security is provided by Apache Knox Gateway. For remote clients, this means that the available cluster services are only accessible by way of secure endpoints.  In this article, I will walk through how to submit Spark jobs to a Data Hub cluster from a remote client, and I will also provide examples demonstrating how to use the API. Apache Spark provides a spark-submit script, which is used to launch applications on a cluster. The spark-submit script depends on having network access to the cluster manager (YARN). A remote client is one that is not deployed on a target Data Hub cluster node, such as a Gateway node. T

In Kafka, topic/partition data is stored in data directories. These directory locations are configured using "log.dirs" config property. We can configure one (or) more directory locations. Kafka balances the partition data directories across these given directory locations. Normally we start with one directory location. With in increase in data size, we may need to add more disks. We can append new directory location to existing "log.dirs" config property. After server restart, Kafka uses new directory location for new partitions. Kafka does not automatically move existing partitions directories to new directory locations. (i.e) It does not auto balance the partitions across directory locations. Some times we want to move some partition data to different location. We have two approaches for this Approach 1: Just delete existing data directory contents and configure new data directories locations In this approach, Kafka replicates the partition data from other members of the cluster

Some people use the Boto3 library to browse their Amazon bucket from Python, and I was searching the same for Azure. This is far from being optimized, but it could be a starting point. First things first, we need to find the Azure access token. Notice those keys are supposed to rotate so you have to have that in mind. In Azure portal, let's get to the Storage Account in the ResourceGroup defined for your account, and click on Access keys There are two keys (for rotation without interruption), let's copy the first one.    In my CML project, I'm defining an AZURE_STORAGE_TOKEN environment variable with that key:   As you see above, 'STORAGE' variable has been populated. If you want it to be automatically populated, here's some code:   !pip3 install git+https://github.com/fletchjeff/cmlbootstrap#egg=cmlbootstrapfrom cmlbootstrap import CMLBootstrap# Instantiate API Wrappercml = CMLBootstrap()# Set the STORAGE environment variabletry : storage=os.environ["STORAGE"

In this article, we will learn to pass atlas-application.properties configuration file from a different location in spark-submit command. When Atlas service is enabled in CDP, and we run Spark application by default, atlas-application.properties file is picked from /etc/spark/conf.cloudera.spark_on_yarn/ directory. Let's test with SparkPi example:   spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode client /opt/cloudera/parcels/CDH/jars/spark-examples*.jar 10   We can see the following output in the application log.   21/08/23 06:12:03 INFO atlas.ApplicationProperties: Looking for atlas-application.properties in classpath21/08/23 06:12:03 INFO atlas.ApplicationProperties: Loading atlas-application.properties from file:/etc/spark/conf.cloudera.spark_on_yarn/atlas-application.properties   If we want to pass the atlas-application.properties configuration file from a different location, for example /tmp directo

Some NiFi Expression Language (EL) expressions can be fairly complex, or used in a large flow, or both. These can make it difficult to test an EL expression on a running NiFi system. Although an excellent feature of NiFi is being able to adapt the flow while the system is running, it may not be prudent to stop a downstream processor, reroute a connection to something like UpdateAttribute, then list the queue in order to see attributes, content, etc.To make EL testing easier, I wrote a Groovy script called testEL.groovy that uses the same EL library that NiFi does, so all functions present in the specified NiFi version are available to the test tool. The following is the usage:usage: groovy testEL.groovy [options] [expressions] Options: -D <attribute=value> set value for given attribute -help print this messageAs an example, the following tests an expression that appends "_world" to the "filename" attribute:> groovy testEL.groovy -D filename=hello '${filena

This video describes a way to hide the plain text passwords in Knox topology. The passwords are basically the Bind User passwords when used in ShiroProvider and LDAP/AD authentication.  

Use Case: Ingesting energy data and running an Apache Spark job as part of the flow.We will be using the new (in Apache NiFi 1.5 / HDF 3.1) ExecuteSparkInteractive processor with the LivyController to accomplish that integration. As we mentioned in the first part of the article, it's pretty easy to set this up.Since this is a modern Apache NiFi project, we use version control on our code:On a local machine, I am talking to an electricity sensor over WiFi in a Python script. This code is processed, cleaned and sent to a cloud hosted Apache NiFi instance via S2S over HTTP.In the cloud we receive the pushed messages.Once we open the Spark It Up processor group, we have a flow to process the data.Flow OverviewQueryRecord: Determine how to route based on query on streaming data. Converts JSON to Apache AVRO.Path for All FilesUpdateAttribute: Set a schemaMergeContent: Do an Apache AVRO merge on our data to make bigger files.ConvertAvroToORC: Build an Apache ORC file from merged

Following is the criteria: ORC and Managed table ACID enabled HDFS directory should be owned by Hive According to the shouldTableBeExternal method in HiveStrictMigration.java, a table is external if it is a StorageHandler table, or if it is an Avro/Text/Parquet table, or if it is a list bucketed table, or the directory is not owned by Hive.    Following is the codebase that describes the same: String reason = shouldTableBeExternal(tableObj, ownerName, conf, hms, isPathOwnedByHive); if (reason != null) { LOG.debug("Converting {} to external table. {}", getQualifiedName(tableObj), reason); result = TableMigrationOption.EXTERNAL; } else { result = TableMigrationOption.MANAGED; Reference: HiveStrictManagedMigration (GitHub) It should be converted to an EXTERNAL table, in case it does not fit the criteria for the upgrade.

This video explains feasible and efficient ways to troubleshoot performance or perform root-cause analysis on any Spark streaming application, which usually tend to grow over the gigabyte size. However, this article does not cover yarn-client mode as it is recommended to use yarn-cluster for streaming applications due to reasons that will not be discussed on this article.   Open the video on YouTube here   Spark streaming applications usually run for long periods of time, before facing issues that may cause them to be shut down. In other cases, the application will not even be shut down, but it could be facing performance degradation during certain peak hours. In any case, the amount and size of this log will keep growing over time, making it really difficult to analyze when they start growing past the gigabyte size.It's well known that Spark, as many other applications, uses log4j facility to handle logs for both the driver and the executors, hence it is recommended to tun

It may often come to pass that you need to utilize data that does not exist in your Environment's Data Lake. This article will cover 2 scenarios where you need to access data that is outside of your Data Lake (either in a Storage Account in the same resource group as your CDP Public Cloud or in a different resource group, albeit in the same subscription). UPDATE: Scenarios 3 and 4 below cover RAZ! Scenario 1 - Adding External Storage Account Access to your Data Access Identity (non RAZ) In this scenario, you may want a single managed identity to be able to access both your Data Lake Storage Account as well as your External Storage Account. For this, I will choose to grant access to only a single container within the storage account (but, as you will see in the following scenario, you could grant access to the entire storage account - I'm showing both purely for illustration). Find your DataAccessIdentity in the Azure Portal, and note it's Object ID Find your External Storage Account

Summary of this article This article attempts to explain the data sources of the Overview and Usage tabs and their meanings in the Admin page of CDSW. Background I have encountered several customers asking me the same questions. It is how to evaluate the resource usage of the CDSW cluster. Below, I will record some of the specific questions asked by customers with a representative meaning, so that I can respond to customers more quickly in the future, and for other people's reference. Why is the usage graphs I observed on Grafana of CDSW not consistent with the resource usage rate observed on the host chart s on CM? For your confusion, from the conclusion: On the CM Web UI, find the page of the specific host. The charts you see under the Chart Library are all accurate. Please use this as a benchmark to judge the true situation of the host's resource usage. The data source for this part is the Host Monitor of Cloudera Management Service, a monitoring service designed and developed by

As per HIVE-12192, by default, Hive displays timestamp in UTC. This article lists the Hive query commands that can be used to change the timestamp. To display the current timestamp in Asia/Kolkata:   SELECT from_utc_timestamp(cast(from_unixtime(cast(current_timestamp() AS bigint)) as TIMESTAMP),"Asia/Kolkata")   To display a UNIX epoch number in Asia/Kolkata:    SELECT from_utc_timestamp(cast(from_unixtime(cast(1623943533 AS bigint)) as TIMESTAMP),"Asia/Kolkata") ;  

To list all the tables that are ACID(TRANSACTIONAL)/ NON_TRANSACTIONAL, perform the following steps: Connect to the backend database Execute "use hive" as all the tables are in the Hive databaseNote: Please check if the database name is different. This is specified at the time of creating the Hive service. Execute the following to check for ACID table:select PARAM_KEY,PARAM_VALUE,TBL_NAME from TABLE_PARAMS,TBLS where PARAM_KEY='transactional' and PARAM_VALUE='true' and TABLE_PARAMS.TBL_ID=TBLS.TBL_ID;​ OR, execute below to check for NON-TRANSACTIONAL  table. select PARAM_KEY,PARAM_VALUE,TBL_NAME from TABLE_PARAMS,TBLS where PARAM_KEY='transactional' and PARAM_VALUE='false' and TABLE_PARAMS.TBL_ID=TBLS.TBL_ID;​    

Note: Cloudera does not support antivirus software of any kind. This article contains recommendations for excluding HDP components and directories from AV scans and monitoring. It is important to note that these recommendations do not apply to each service, and further, some services will have additional items to exclude which are unique to them. These details will be addressed in individual articles dedicated to the service in question. The three primary locations you will want to exclude from Antivirus are: Data directories: These can be very large, and therefore take a long time to scan; they can also be very write-heavy, and therefore suffer performance impacts or failures if the AV holds up writes. Log directories: These are write-heavy. Scratch directories: These are internal locations used by some services for writing temporary data, and can also cause performance impacts or failures if the AV holds up writes. In general, consider excluding the following directories and all o

When a Hive replication policy executes in Replication Manager, the admin can monitor the details of the policy and its execution by checking the underlying command details on the source CDH Cluster's CM. In order to export the data and metadata related to the specified Hive tables, multiple steps are executed in the source Cloudera Manager. The Export Hive Metastore section has multiple steps that are explained below:   The first subcommand exports the Hive Metadata for the input database and tables. The next step transfers the HDFS files from the CDH cluster to the cloud storage, which is used by the Data Lake cluster. The next step exports the relevant Sentry permissions. Once the above steps are executed successfully, the workflow resumes on the Data Lake cluster as explained in the previous section.  With these steps, the administration now has a detailed behind-the-scenes view of the steps involved in running a Hive replication policy.