Spark writes dataframe data to the Hive partition table

From spark1.2 to spark1.3, spark SQL in the SchemaRDD into a DataFrame, DataFrame relative to the SchemaRDD has been greatly changed, while providing more useful and convenient API.
DataFrame data will be written to hive, the default is hive default database, insertInto does not specify the parameters of the database, this article uses the following way to write data to the hive table or hive table partition, for reference only.
1, the DataFrame data written to the Hive table from the DataFrame class can be seen with the hive table related to the Api write the following:

RegisterTempTable (tableName: String): Unit,
InsertInto (tableName: String): Unit
InsertInto (tableName: String, overwrite: Boolean): Unit
SaveAsTable (tableName: String, source: String, mode: SaveMode, options: Map [String, String]): Unit

There are many overloaded functions that are not enumerated
The registerTempTable function is created by the spark temporary table
InsertInto function is to write data to the table, you can see that this function can not specify the database and partition information, can not be directly written.
Write data to the hive data warehouse must specify the database, hive data table can be established in the hive, or use hiveContext.sql ("create table ....")
The following statement writes data to the specified database data table:
1
2
3
4
5
6
7
case class Person( name :String,col1: Int ,col2:String)
val sc = new org.apache.spark.SparkContext
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
import hiveContext.implicits._
hiveContext.sql( "use DataBaseName" )
).map(x=>x.split( "\\s+" )).map(x=>Person(x(0),x(1).toInt,x(2)))
data.toDF().insertInto( "tableName" )
Create a case class will RDD data type into the case class type, and then through toDF converted to DataFrame, call insertInto function, first specify the database, using the hiveContext.sql ("use DataBaseName") statement, you can DataFrame data Write in the hive data table

2, the DataFrame data written to the hive designated data table partition
Hive data set can be established in the hive, or use hiveContext.sql ("create table ...."), the use of saveAsTable data storage format is limited, the default format for the parquet, can be specified as json, if there are other formats specified, Try to use the statement to create the hive table.
Write the data to the partition table The idea is to first write the DataFrame data to a temporary table, followed by the hiveContext.sql statement to write the data to the hive partition table. The specific operation is as follows:
1
2
3
4
5
6
7
8
case class Person( name :String,col1: Int ,col2:String)
val sc = new org.apache.spark.SparkContext
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
import hiveContext.implicits._
hiveContext.sql( "use DataBaseName" )
).map(x=>x.split( "\\s+" )).map(x=>Person(x(0),x(1).toInt,x(2)))
data.toDF().registerTempTable( "table1" )
hiveContext.sql( "insert into table2 partition(date='2015-04-02') select name,col1,col2 from table1" )
Use the above method to write dataframe data to the hive partition table

Commentaires

Enregistrer un commentaire

Posts les plus consultés de ce blog

Controlling Parallelism in Spark by controlling the input partitions by controlling the input partitions

Image
It is useful to be able to control the degree of parallelism when using Spark. Spark provides a very convenient method to increase the degree of parallelism which should be adequate in practice. This blog entry of the "Inside Spark" series describes the knobs Spark uses to control the degree of parallelism. Controlling the number of partitions in the " local " mode Spark is designed ground up to enable unit testing. To control the degree of parallelism in the local mode simply utilize the function JavaRDD<String> org.apache.spark.api.java.JavaSparkContext.parallelize(List<String> list, int numSlices) The following snippet of code demonstrates this usage- public class SparkControlPartitionSizeLocally { public static void main ( String [] args ) { JavaSparkContext sc = new JavaSparkContext ( "local" , "localpartitionsizecontrol" ); String input = "four score and sev...
Lire la suite

Spark performance optimization: shuffle tuning

Image
Article directory 1 shuffle tuning 1.1 Summary of tuning 1.2 Overview of ShuffleManager Development 1.3 HashShuffleManager operating principle 1.3.1 Unoptured HashShuffleManager 1.3.2 Optimized HashShuffleManager 1.4 SortShuffleManager operating principle 1.4.1 General operating mechanism 1.4.2 bypass running mechanism 1.5 shuffle related parameters tuning 1.5.1 spark.shuffle.file.buffer 1.5.2 spark.reducer.maxSizeInFlight 1.5.3 spark.shuffle.io.maxRetries 1.5.4 spark.shuffle.io.retryWait 1.5.5 spark.shuffle.memoryFraction 1.5.6 spark.shuffle.manager 1.5.7 spark.shuffle.sort.bypassMergeThreshold 1.5.8 spark.shuffle.consolidateFiles 2 write in the last words Shuffle Summary of tuning Most of the performance of Spark operations is mainly consumed in the shuffle link, because the link contains a large number of disk IO, serialization, network data transmission and othe...
Lire la suite

Spark optimization

Image
One operation and maintenance 1. Master hang up, standby restart is also invalid Master defaults to 512M of memory, when the task in the cluster is particularly high, it will hang, because the master will read each task event log log to generate spark ui, the memory will naturally OOM, you can run the log See that the master of the start through the HA will naturally fail for this reason. solve Increase the Master's memory spark-env.sh , set in the master node spark-env.sh : export SPARK_DAEMON_MEMORY 10g # 根据你的实际情况 Reduce the job information stored in the Master memory spark.ui.retainedJobs 500 # 默认都是1000 spark.ui.retainedStages 500 Hang up or suspend Sometimes we will see the web node in the web ui disappear or in the dead state, the task of running the node will report a variety of lost worker errors, causing the same reasons and the above, worker memory to save a lot of ui The information leads to gc when the heartbeat is lost...
Lire la suite