Hive 动态分区 - Spark 的并发写入会损坏数据

Question

我们已经设置了一个 Spark 作业来插入到 Hive 中（使用数据帧）。hive表设置为创建动态分区。只要我们运行一个 Spark 作业将数据插入 Hive，一切都会完美运行。

我们遇到的问题是，我们预计将运行并发 Spark 作业以将数据加载到 Hive 中。这似乎不起作用。我读到动态分区不提供独占锁，而是提供共享锁。在我们的例子中，我们可以看到，如果同时运行 4-5 个 Spark 作业，数据就会损坏，一些记录会丢失。非常容易重现，几乎每次都会发生。

有人解决这个问题了吗？即使用并发作业的动态分区插入到配置单元表中，并且仍然确保不会发生数据损坏。任何意见都非常感谢！

火花代码片段：

// Set hive conf to allow dynamic partitions to be created
sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")

//Create temp table to load data into Hive
parsedDataDF.registerTempTable("parsedDatatempTable")

//Insert data into Hive, dynamic partitioning   
sqlContext.sql("insert into table " + hiveDBToLoad + "." + hiveTableToLoad + " partition (partition_1, partition_2, partition_3) " +
    "select * from parsedDatatempTable")

Answer 1

    $partition_tbl=array();
$partition_tbl[]=array(
    'tbl'=>$wpdb->prefix.'postmeta',
    'field'=>'meta_id',
    'limit'=>2,
    'part_key'=>'p',
);
$partition_tbl[]=array(
    'tbl'=>$wpdb->prefix.'posts',
    'field'=>'ID',
    'limit'=>2,
    'part_key'=>'pos',
);
        
if(!empty($partition_tbl)){
    foreach($partition_tbl as $ppptt){
        $field=$ppptt['field'];
        $tbl=$ppptt['tbl'];
        $limit=$ppptt['limit'];
        $part_key=$ppptt['part_key'];
        $get_max = $wpdb->get_results("SELECT * FROM ".$wpdb->prefix."postmeta ORDER BY $field DESC LIMIT 0, 1");
        if(!empty($get_max)){
            $max=$get_max[0]->$field;
            if($max > $limit){                              
                $partcc=ceil($max/$limit);          
                $has_partitions = $wpdb->get_results("EXPLAIN partitions SELECT * FROM ".$tbl);     
                $haspartarr=array('none');
                if(!empty($has_partitions)){
                    if(!empty($has_partitions[0]->partitions)){
                        $haspartarr=explode(",",$has_partitions[0]->partitions);
                        if(count($haspartarr) > 1){
                             unset($haspartarr[count($haspartarr)-1]);
                        }
                    }
                }
                $part_sql='';
                $part_arr=array();
                $first=true;
                for($i=0;$i<$partcc;$i++){
                    $cpart=$part_key.$i;
                    $reclim=$limit*($i+1);
                    if(!in_array($cpart,$haspartarr)){
                        if(empty($has_partitions[0]->partitions)){
                            $part_arr[]="PARTITION $cpart VALUES LESS THAN ($reclim)";
                        }else{
                            $nwlmn=($reclim-$limit);                        
                            if($first){
                                $part_sql.="ALTER TABLE $tbl REORGANIZE PARTITION $cpart INTO (";
                                $first=false;
                                $morel=$nwlmn-$limit;                                       
                            }
                            $part_arr[]="PARTITION $cpart VALUES LESS THAN ($reclim)";                                          
                        }
                    }
                }
                if(empty($has_partitions[0]->partitions) && !empty($part_arr)){
                    $cpart=$part_key.$i;                
                    $part_arr[]="PARTITION $cpart VALUES LESS THAN MAXVALUE";               
                    $part_sql.="ALTER TABLE $tbl PARTITION BY RANGE($field)(";
                    $part_sql.=implode(",",$part_arr);
                    $part_sql.=")";
                    $wpdb->get_results($part_sql);
                }else{
                    if(!empty($part_arr)){
                        $cpart=$part_key.$i;
                        $part_arr[]="PARTITION $cpart VALUES LESS THAN MAXVALUE";
                        $part_sql.=implode(",",$part_arr);      
                        $part_sql.=");";
                        $wpdb->get_results($part_sql);
                    }           
                }
            }
        }

    }
}