I'm trying to do a algorithm in Spark Mllib. I'm trying to do a Market Basket Analysis. And I've as a data set this:  ID bigint,  Chain int,  Dept int,  Category int,  Company bigint,  Brand bigint,  Date string,  Product_Size double,  Product_Measure string,  Purchase_Quantity int,   Purchase_Amount doubleThe code that I found is this:import org.apache.spark.mllib.fpm.FPGrowthimport org.apache.spark.rdd.RDDimport sys.process._val data = sc.textFile(/user/admin/retail/marketbaskets/part-r-00000)val transactions: RDD[Array[String]] = data.map(s => s.trim.split(','))val fpg = new FPGrowth()  .setMinSupport(0.007)  .setNumPartitions(10)val model = fpg.run(transactions)model.freqItemsets.collect().foreach { itemset =>  println(itemset.items.mkString([, ,, ]) + ,  + itemset.freq)}val minConfidence = 0.8model.generateAssociationRules(minConfidence).collect().foreach { rule =>  println(    rule.antecedent.mkString([, ,, ])      +  =>  + rule.consequent .mkString([, ,, ])      + ,  + rule.confidence)}However when I submit this line of code:val model = fpg.run(transactions)I get the fowllowing error:val model = fpg.run(transactions)16/08/26 10:56:09 WARN fpm.FPGrowth: Input data is not cached.16/08/26 10:56:21 ERROR executor.Executor: Exception in task 0.0 in stage 10.0 (TID 7)org.apache.spark.SparkException: Items in a transaction must be unique but got WrappedArray(13873775, 4, 99, 9909, 102113020, 15704, 2012-03-19:00, 6.25, OZ, 4, 11.96).    at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:143) at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:140) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:189) at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) at org.apache.spark.scheduler.Task.run(Task.scala:89) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745)16/08/26 10:56:21 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 10.0 (TID 7, localhost): org.apache.spark.SparkException: Items in a transaction must be unique but got WrappedArray(13873775, 4, 99, 9909, 102113020, 15704, 2012-03-19:00, 6.25, OZ, 4, 11.96). at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:143) at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:140)    at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:189) at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) at org.apache.spark.scheduler.Task.run(Task.scala:89) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)    at java.lang.Thread.run(Thread.java:745)16/08/26 10:56:21 ERROR scheduler.TaskSetManager: Task 0 in stage 10.0 failed 1 times; aborting joborg.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 10.0 failed 1 times, most recent failure: Lost task 0.0 in stage 10.0 (TID 7, localhost): org.apache.spark.SparkException: Items in a transaction must be unique but got WrappedArray(13873775, 4, 99, 9909, 102113020, 15704, 2012-03-19:00, 6.25, OZ, 4, 11.96).    at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:143) at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:140) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:189)    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64)    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)    at org.apache.spark.scheduler.Task.run(Task.scala:89)    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)    at java.lang.Thread.run(Thread.java:745)Driver stacktrace:    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1431) at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1419)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1418) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1418) at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799) at scala.Option.foreach(Option.scala:236) at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:799) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1640) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1599) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1588) at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:620) at org.apache.spark.SparkContext.runJob(SparkContext.scala:1843) at org.apache.spark.SparkContext.runJob(SparkContext.scala:1856) at org.apache.spark.SparkContext.runJob(SparkContext.scala:1869) at org.apache.spark.SparkContext.runJob(SparkContext.scala:1940) at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:927) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111) at org.apache.spark.rdd.RDD.withScope(RDD.scala:316) at org.apache.spark.rdd.RDD.collect(RDD.scala:926) at org.apache.spark.mllib.fpm.FPGrowth.genFreqItems(FPGrowth.scala:149) at org.apache.spark.mllib.fpm.FPGrowth.run(FPGrowth.scala:118) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:38) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:43) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:45)    at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:47) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:49) at $iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:51) at $iwC$$iwC$$iwC$$iwC.<init>(<console>:53)    at $iwC$$iwC$$iwC.<init>(<console>:55) at $iwC$$iwC.<init>(<console>:57) at $iwC.<init>(<console>:59) at <init>(<console>:61) at .<init>(<console>:65) at .<clinit>(<console>) at .<init>(<console>:7) at .<clinit>(<console>) at $print(<console>) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1045) at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1326) at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:821) at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:852) at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:800) at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:857) at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:902) at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:814) at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:657) at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:665) at org.apache.spark.repl.SparkILoop.org$apache$spark$repl$SparkILoop$$loop(SparkILoop.scala:670)    at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply$mcZ$sp(SparkILoop.scala:997) at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:945) at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:945)    at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135) at org.apache.spark.repl.SparkILoop.org$apache$spark$repl$SparkILoop$$process(SparkILoop.scala:945) at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:1064) at org.apache.spark.repl.Main$.main(Main.scala:31)    at org.apache.spark.repl.Main.main(Main.scala)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:606)    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731) at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181) at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121) at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)Caused by: org.apache.spark.SparkException: Items in a transaction must be unique but got WrappedArray(13873775, 4, 99, 9909, 102113020, 15704, 2012-03-19:00, 6.25, OZ, 4, 11.96). at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:143)    at org.apache.spark.mllib.fpm.FPGrowth$$anonfun$1.apply(FPGrowth.scala:140) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:189)    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64)    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)    at org.apache.spark.scheduler.Task.run(Task.scala:89)    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)    at java.lang.Thread.run(Thread.java:745)I already try to use this:val model = fpg.run(transactions.values.map(_.toArray))But it still gives me an error. I also see in the Hive if this error was a duplicate case but it isn't.Anyone knows how can I solve this error?Many thanks!
Items in a transaction must be unique but got WrappedArray
apache spark;scala;market basket analysis
org.apache.spark.SparkException: Items in a transaction must be unique but got WrappedArray(13873775, 4, 99, 9909, 102113020, 15704, 2012-03-19:00, 6.25, OZ, 4, 11.96).The error pretty much tells you the problem. A basket is a set of unique items, and 4 occurs twice here. That's not valid input.However you have a much bigger problem here. You are parsing a line of text that clearly is just a description of one item, and interpreting it as a bunch of item IDs. You have a date, weight, units, etc. This is completely invalid as input.