Hello my 2 cents here, hope it helps If you want to just to play around with Spark, i'd leave Hadoop out, it's an unnecessary dependency that you dont need for just running a python script Instead do the following: - got to the root of our master / slave node. create a directory /root/pyscripts - place your csv file there as well as the python script - run the script to replicate the whole directory across the cluster (i believe it's called copy-script.sh) - then run your spark-submit , it will be something lke ./spark-submit /root/pyscripts/mysparkscripts.py file:///root/pyscripts/tree_addhealth.csv 10 --master spark://ip-172-31-44-155.us-west-2.compute.internal:7077 - in your python script, as part of your processing, write the parquet file in directory /root/pyscripts
If you have an AWS account and you are versatile with that - you need to setup bucket permissions etc - , you can just - store your file in one of your S3 bucket - create an EMR cluster - connect to master or slave - run your scritp that reads from the s3 bucket and write to the same s3 bucket Feel free to mail me privately, i have a working script i have used to test some code on spark standalone cluster hth On Thu, Aug 3, 2017 at 10:30 AM, Gourav Sengupta <gourav.sengu...@gmail.com> wrote: > Hi Steve, > > I love you mate, thanks a ton once again for ACTUALLY RESPONDING. > > I am now going through the documentation (https://github.com/ > steveloughran/hadoop/blob/s3guard/HADOOP-13786- > committer/hadoop-tools/hadoop-aws/src/site/markdown/tools/ > hadoop-aws/s3a_committer_architecture.md) and it makes much much more > sense now. > > Regards, > Gourav Sengupta > > On Thu, Aug 3, 2017 at 10:09 AM, Steve Loughran <ste...@hortonworks.com> > wrote: > >> >> On 2 Aug 2017, at 20:05, Gourav Sengupta <gourav.sengu...@gmail.com> >> wrote: >> >> Hi Steve, >> >> I have written a sincere note of apology to everyone in a separate email. >> I sincerely request your kind forgiveness before hand if anything does >> sound impolite in my emails, in advance. >> >> Let me first start by thanking you. >> >> I know it looks like I formed all my opinion based on that document, but >> that is not the case at all. If you or anyone tries to execute the code >> that I have given then they will see what I mean. Code speaks louder and >> better than words for me. >> >> So I am not saying you are wrong. I am asking verify and expecting >> someone will be able to correct a set of understanding that a moron like >> me has gained after long hours of not having anything better to do. >> >> >> SCENARIO: there are two files file1.csv and file2.csv stored in HDFS with >> replication 2 and there is a HADOOP cluster of three nodes. All these nodes >> have SPARK workers (executors) running in them. Both are stored in the >> following way: >> ----------------------------------------------------- >> | SYSTEM 1 | SYSTEM 2 | SYSTEM 3 | >> | (worker1) | (worker2) | (worker3) | >> | (master) | | | >> ----------------------------------------------------- >> | file1.csv | | file1.csv | >> ----------------------------------------------------- >> | | file2.csv | file2.csv | >> ----------------------------------------------------- >> | file3.csv | file3.csv | | >> ----------------------------------------------------- >> >> >> >> >> >> CONSIDERATION BASED ON WHICH ABOVE SCENARIO HAS BEEN DRAWN: >> HDFS replication does not store the same file in all the nodes in the >> cluster. So if I have three nodes and the replication is two then the same >> file will be stored physically in two nodes in the cluster. Does that sound >> right? >> >> >> HDFS breaks files up into blocks (default = 128MB). If a .csv file is > >> 128 then it will be broken up into blocks >> >> file1.cvs -> [block0001, block002, block0003] >> >> and each block will be replicated. With replication = 2 there will be two >> copies of each block, but the file itself can span > 2 hosts. >> >> >> ASSUMPTION (STEVE PLEASE CLARIFY THIS): >> If SPARK is trying to process to the records then I am expecting that >> WORKER2 should not be processing file1.csv, and similary WORKER 1 should >> not be processing file2.csv and WORKER3 should not be processing file3.csv. >> Because in case WORKER2 was trying to process file1.csv then it will >> actually causing network transmission of the file unnecessarily. >> >> >> Spark prefers to schedule work locally, so as to save on network traffic, >> but it schedules for execution time over waiting for workers free on the >> node with the data. IF a block is on nodes 2 and 3 but there is only a free >> thread on node 1, then node 1 gets the work >> >> There's details on whether/how work across blocks takes place which I'm >> avoiding. For now know those formats which are "splittable" will have work >> scheduled by block. If you use Parquet/ORC/avro for your data and compress >> with snappy, it will be split. This gives you maximum performance as >1 >> thread can work on different blocks. That is, if file1 is split into three >> blocks, three worker threads can process it. >> >> >> ASSUMPTION BASED ON ABOVE ASSUMPTION (STEVE ONCE AGAIN, PLEASE CLARIFY >> THIS): >> if WORKER 2 is not processing file1.csv then how does it matter whether >> the file is there or not at all in the system? Should not SPARK just ask >> the workers to process the files which are avialable in the worker nodes? >> In case both WORKER2 and WORKER3 fails and are not available then file2.csv >> will not be processed at all. >> >> >> locality is best-effort, not guaranteed. >> >> >> ALSO I DID POST THE CODE AND I GENUINELY THINK THAT THE CODE SHOULD BE >> EXECUTED (Its been pointed out that I am learning SPARK, and even I did not >> take more than 13 mins to set up the cluster and run the code). >> >> Once you execute the code then you will find that: >> 1. if the path starts with file:/// while reading back then there is no >> error reported, but the number of records reported back are only those >> records in the worker which also has the server. >> 2. also you will notice that once you cache the file before writing the >> partitions are ditributed nicely across the workers, and while writing >> back, the dataframe partitions does write properly to the worker node in >> the Master, but the workers in the other system have the files written in >> _temporary folder which does not get copied back to the main folder. >> Inspite of this the job is not reported as failed in SPARK. >> >> >> This gets into the "commit protocol". You don't want to know all the >> dirty details (*) but essentially its this >> >> 1. Every worker writes its output to a directory under the destination >> directory, something like '$dest/_temporary/$appAtt >> emptId/_temporary/$taskAttemptID' >> 2. it is the spark driver which "commits" the job by moving the output >> from the individual workers from the temporary directories into $dest, then >> deleting $dest/_temporary >> 3. For which it needs to be able to list all the output in >> $dest/_temporary >> >> In your case, only the output on the same node of the driver is being >> committed, because only those files can be listed and moved. The output on >> the other nodes isn't seen, so isn't committed, nor cleaned up. >> >> >> >> Now in my own world, if I see, the following things are happening, >> something is going wrong (with me): >> 1. SPARK transfers files from different systems to process, instead of >> processing them locally (I do not have code to prove this, and therefore >> its just an assumption) >> 2. SPARK cannot determine when the writes are failing in standalone >> clusters workers and reports success (code is there for this) >> 3. SPARK reports back number of records in the worker running in the >> master node when count() is given without reporting an error while using >> file:/// and reports an error when I mention the path without file:/// >> (for SPARK 2.1.x onwards, code is there for this) >> >> >> >> s everyone's been saying, file:// requires a shared filestore, with >> uniform paths everywhere. That's needed to list the files to process, read >> the files in the workers and commit the final output. NFS cross-mounting is >> the simplest way to do this, especially as for three nodes HDFS is >> overkill: more services to keep running, no real fault tolerance. Export a >> directory tree from one of the servers, give the rest access to it, don't >> worry about bandwidth use as the shared disk itself will become the >> bottleneck >> >> >> >> I very sincerely hope with your genuine help the bar of language and >> social skills will be lowered for me. And everyone will find a way to >> excuse me and not qualify this email as a means to measure my extremely >> versatile and amazingly vivid social skills. It will be a lot of help to >> just focus on the facts related to machines, data, error and (the language >> that I somehow understand better) code. >> >> >> My sincere apologies once again, as I am 100% sure that I did not meet >> the required social and language skills. >> >> Thanks a ton once again for your kindness, patience and understanding. >> >> >> Regards, >> Gourav Sengupta >> >> >> * for the curious, the details of the v1 and v2 commit protocols are >> https://github.com/steveloughran/hadoop/blob/s3guard/HADOOP- >> 13786-committer/hadoop-tools/hadoop-aws/src/site/markdown/ >> tools/hadoop-aws/s3a_committer_architecture.md >> >> Like I said: you don't want to know the details, and you really don't >> want to step through Hadoop's FileOutputCommitter to see what's going on. >> The Spark side is much easier to follow. >> >> >