Coding Zone

Just to cover more following steps after kicking off the query: INSERT OVERWRITE LOCAL DIRECTORY '/home/lvermeer/temp' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' select books from table;

In my case, the generated data under temp folder is in deflate format, and it looks like this:

$ ls 000000_0.deflate   000001_0.deflate   000002_0.deflate   000003_0.deflate   000004_0.deflate   000005_0.deflate   000006_0.deflate   000007_0.deflate

Here's the command to unzip the deflate files and put everything into one csv file:

How to transform a column of json strings to structs

....

With Spark 2, we have seen this occur when users have missed adding a new (coming from Spark 1.6) method call into their SparkSession initialization. See further details in the Spark documentation.

The resolution is to include a call to enableHiveSupport():

Assuming, you are using Spark 2.1.0 or later and my_DF is your dataframe,

//get the schema split as string with comma-separated field-datatype pairs StructType my_schema = my_DF.schema(); StructField[] fields = my_schema.fields(); String fieldStr = ""; for (StructField f : fields) { fieldStr += f.name() + " " + f.dataType().typeName() + ","; } //drop the table if already created spark.sql("drop table if exists my_table"); //create the table using the dataframe schema spark.sql("create table my_table(" + fieldStr.subString(0,fieldStr.length()-1)+ ") row format delimited fields terminated by '|' location '/my/hdfs/location'");    //write the dataframe data to the hdfs location for the created Hive table    my_DF.write()    .format("com.databricks.spark.csv")    .option("delimiter","|")    .mode("overwrite")    .save("/my/hdfs/location");

The other method using temp table

Question is somewhat unclear, but if you're looking for a way to "flatten" a DataFrame schema (i.e. get an array of all non-struct fields), here's one:

def flatten(schema: StructType): Array[StructField] = schema.fields.flatMap { f =>  f.dataType match {    case struct: StructType => flatten(struct)    case _ => Array(f)  } }

For example:

https://databricks.com/blog/2015/02/02/an-introduction-to-json-support-in-spark-sql.html

// Create a SQLContext (sc is an existing SparkContext) val sqlContext = new org.apache.spark.sql.SQLContext(sc) // Suppose that you have a text file called people with the following content: // {"name":"Yin", "address":{"city":"Columbus","state":"Ohio"}} // {"name":"Michael", "address":{"city":null, "state":"California"}} // Create a SchemaRDD for the JSON dataset. val people = sqlContext.jsonFile("[the path to file people]") // Register the created SchemaRDD as a temporary table. people.registerTempTable("people")

scala> val people = sqlContext.jsonFile("test/person.json") warning: there was one deprecation warning; re-run with -deprecation for details people: org.apache.spark.sql.DataFrame = [address: struct<city: string, state: string>, name: string] people.printSchema() root |-- address: struct (nullable = true) | |-- city: string (nullable = true) | |-- state: string (nullable = true) |-- name: string (nullable = true)

Apache Spark's classpath is built dynamically (to accommodate per-application user code) which makes it vulnerable to such issues. @user7337271's answer is correct, but there are some more concerns, depending on the cluster manager ("master") you're using.

First, a Spark application consists of these components (each one is a separate JVM, therefore potentially contains different classes in its classpath):

Driver: that's your application creating a SparkSession (or SparkContext) and connecting to a cluster manager to perform the actual work

Cluster Mamanger: serves as an "entry point" to the cluster, in charge of allocating executorsfor each application. There are several different types supported in Spark: standalone, YARN and Mesos, which we'll describe bellow.

Executors: these are the processes on the cluster nodes, performing the actual work (running Spark tasks)

The relationsip between these is described in this diagram from Apache Spark's cluster mode overview:

Now - which classes should reside in each of these components?

This can be answered by the following diagram:

Let's parse that slowly:

Spark Code are Spark's libraries. They should exist in ALL three components as they include the glue that let's Spark perform the communication between them. By the way - Spark authors made a design decision to include code for ALL components in ALL components (e.g. to include code that should only run in Executor in driver too) to simplify this - so Spark's "fat jar" (in versions up to 1.6) or "archive" (in 2.0, details bellow) contain the necessary code for all components and should be available in all of them.

Driver-Only Code this is user code that does not include anything that should be used on Executors, i.e. code that isn't used in any transformations on the RDD / DataFrame / Dataset. This does not necessarily have to be separated from the distributed user code, but it can be.

Distributed Code this is user code that is compiled with driver code, but also has to be executed on the Executors - everything the actual transformations use must be included in this jar(s).

Now that we got that straight, how do we get the classes to load correctly in each component, and what rules should they follow?

Spark Code: as previous answers state, you must use the same Scala and Spark versions in all components.

1.1 In Standalone mode, there's a "pre-existing" Spark installation to which applications (drivers) can connect. That means that all drivers must use that same Spark versionrunning on the master and executors.

1.2 In YARN / Mesos, each application can use a different Spark version, but all components of the same application must use the same one. That means that if you used version X to compile and package your driver application, you should provide the same version when starting the SparkSession (e.g. via spark.yarn.archive or spark.yarn.jars parameters when using YARN). The jars / archive you provide should include all Spark dependencies (including transitive dependencies), and it will be shipped by the cluster manager to each executor when the application starts.

Driver Code: that's entirely up to - driver code can be shipped as a bunch of jars or a "fat jar", as long as it includes all Spark dependencies + all user code

Distributed Code: in addition to being present on the Driver, this code must be shipped to executors (again, along with all of its transitive dependencies). This is done using the spark.jars parameter.

To summarize, here's a suggested approach to building and deploying a Spark Application (in this case - using YARN):

Create a library with your distributed code, package it both as a "regular" jar (with a .pom file describing its dependencies) and as a "fat jar" (with all of its transitive dependencies included).

Create a driver application, with compile-dependencies on your distributed code library and on Apache Spark (with a specific version)

Package the driver application into a fat jar to be deployed to driver

Pass the right version of your distributed code as the value of spark.jars parameter when starting the SparkSession

The recommended technique to convert between Java and Scala collections (since Scala 2.8.1) is via scala.collection.JavaConverters. This gives you more control than JavaConversions and avoids some possible implicit conflicts.

You won't have a direct implicit conversion this way. Instead, you get asScala and asJavamethods pimped onto collections, allowing you to perform the conversions explicitly.

To convert a Java iterator to a Scala iterator:

javaIterator.asScala

To convert a Java iterator to a Scala List (via the scala iterator):

javaIterator.asScala.toList

You may also want to consider converting toSeq instead of toList. In the case of iterators, this'll return a Stream - allowing you to retain the lazy behaviour of iterators within the richer Seqinterface.

EDIT: There's no toVector method, but (as Daniel pointed out) there's a toIndexedSeq method that will return a Vector as the default IndexedSeq subclass (just as List is the default Seq).

-bash-4.2$ cat /etc/*rele* NAME="Red Hat Enterprise Linux Server" VERSION="7.3 (Maipo)" ID="rhel" ID_LIKE="fedora" VERSION_ID="7.3" PRETTY_NAME="Red Hat Enterprise Linux Server 7.3 (Maipo)" ANSI_COLOR="0;31" CPE_NAME="cpe:/o:redhat:enterprise_linux:7.3:GA:server" HOME_URL="https://www.redhat.com/" BUG_REPORT_URL="https://bugzilla.redhat.com/"

1. find out the path of the hive tables:

        for example, find the path for table r_scan1, 

        hive> describe formatted r_scan1;

        => Location:           maprfs:/hive/username.db/r_scan1

Then you know the default path is “maprfs:/hive/username.db/”

2. Run the following command

$ hadoop fs -du /hive/username.db/ | awk '/^[0-9]+/ { print int($1/(1024**3)) " [GB]\t" $2 }'

=>

hive -S -e "describe formatted <table_name> ;" | grep 'Location' | awk '{ print $NF }'

https://stackoverflow.com/a/10412663/2345313

hive> set hive.metastore.warehouse.dir;