स्पार्क: उच्चतम टाइमस्टैम्प वाली पंक्ति

Question

रखते हुए डेटाफ्रेम पर ड्रॉप ड्रॉप डुप्लिकेट कैसे करें मेरे पास एक ऐसा केस है जहां मुझे डेटाफ्रेम की डुप्लिकेट पंक्तियां छोड़नी होंगी (इस मामले में डुप्लिकेट का अर्थ है कि उनके पास एक ही 'आईडी' फ़ील्ड है) पंक्ति को उच्चतम 'टाइमस्टैंप' (यूनिक्स टाइमस्टैम्प) फ़ील्ड के साथ रखते हुए।

मुझे drop_duplicate विधि मिली (मैं pyspark का उपयोग कर रहा हूं), लेकिन किसी पर नियंत्रण नहीं है कि किस आइटम को रखा जाएगा।

कोई भी मदद कर सकता है? अग्रिम में Thx

Answer 1

एक मैनुअल मानचित्र और कम कार्यक्षमता जो आप चाहते हैं कार्यक्षमता प्रदान करने के लिए आवश्यक हो सकता है।

def selectRowByTimeStamp(x,y): 
    if x.timestamp > y.timestamp: 
     return x 
    return y 

dataMap = data.map(lambda x: (x.id, x)) 
uniqueData = dataMap.reduceByKey(selectRowByTimeStamp) 

यहां हम आईडी के आधार पर सभी डेटा को समूहीकृत कर रहे हैं। फिर, जब हम समूह को कम कर रहे हैं, तो हम रिकॉर्ड को उच्चतम टाइमस्टैम्प के साथ रखते हुए ऐसा करते हैं। जब कोड कम हो जाता है, तो प्रत्येक आईडी के लिए केवल 1 रिकॉर्ड छोड़ा जाएगा।

Answer 2

आप कुछ इस तरह कर सकते हैं:

val df = Seq(
    (1,12345678,"this is a test"), 
    (1,23456789, "another test"), 
    (2,2345678,"2nd test"), 
    (2,1234567, "2nd another test") 
).toDF("id","timestamp","data") 

+---+---------+----------------+ 
| id|timestamp|   data| 
+---+---------+----------------+ 
| 1| 12345678| this is a test| 
| 1| 23456789| another test| 
| 2| 2345678|  2nd test| 
| 2| 1234567|2nd another test| 
+---+---------+----------------+ 

df.join(
    df.groupBy($"id").agg(max($"timestamp") as "r_timestamp").withColumnRenamed("id", "r_id"), 
    $"id" === $"r_id" && $"timestamp" === $"r_timestamp" 
).drop("r_id").drop("r_timestamp").show 
+---+---------+------------+ 
| id|timestamp|  data| 
+---+---------+------------+ 
| 1| 23456789|another test| 
| 2| 2345678| 2nd test| 
+---+---------+------------+ 

तुम वहाँ एक id के लिए एक दोहराया timestamp हो सकता है (नीचे टिप्पणी देखें) उम्मीद हैं, तो आप ऐसा कर सकते हैं:

df.dropDuplicates(Seq("id", "timestamp")).join(
    df.groupBy($"id").agg(max($"timestamp") as "r_timestamp").withColumnRenamed("id", "r_id"), 
    $"id" === $"r_id" && $"timestamp" === $"r_timestamp" 
).drop("r_id").drop("r_timestamp").show