Re: How can I read ftp
We have solved it using an orchestrator, which copied data from FTP to HDFS. But of course, you can just use a Java FTP client to just read files, put them somewhere, and then read with Spark. пн, 9 авг. 2021 г. в 06:39, Sean Owen : > FTP is definitely not supported. Read the files to distributed storage > first then read from there. > > On Sun, Aug 8, 2021, 10:18 PM igyu wrote: > >> val ftpUrl = "ftp://ftpuser:ftpuser@10.3.87.51:21/sparkftp/"; >> >> val schemas = StructType(List( >> new StructField("name", DataTypes.StringType, true), >> new StructField("age", DataTypes.IntegerType, true), >> new StructField("remk", DataTypes.StringType, true))) >> >>val DF = sparkSession.read.format("csv") >> .schema(schemas) >> .option("header","true") >> .load(ftpUrl) >> // .filter("created<=1602864000") >> >> DF.printSchema() >> DF.show() >> >> I get error >> >> Exception in thread "main" java.lang.IllegalArgumentException: Illegal >> pattern component: XXX >> at >> org.apache.commons.lang3.time.FastDatePrinter.parsePattern(FastDatePrinter.java:282) >> at >> org.apache.commons.lang3.time.FastDatePrinter.init(FastDatePrinter.java:149) >> at >> org.apache.commons.lang3.time.FastDatePrinter.(FastDatePrinter.java:142) >> at >> org.apache.commons.lang3.time.FastDateFormat.(FastDateFormat.java:384) >> at >> org.apache.commons.lang3.time.FastDateFormat.(FastDateFormat.java:369) >> at >> org.apache.commons.lang3.time.FastDateFormat$1.createInstance(FastDateFormat.java:91) >> at >> org.apache.commons.lang3.time.FastDateFormat$1.createInstance(FastDateFormat.java:88) >> at >> org.apache.commons.lang3.time.FormatCache.getInstance(FormatCache.java:82) >> at >> org.apache.commons.lang3.time.FastDateFormat.getInstance(FastDateFormat.java:165) >> at >> org.apache.spark.sql.execution.datasources.csv.CSVOptions.(CSVOptions.scala:139) >> at >> org.apache.spark.sql.execution.datasources.csv.CSVOptions.(CSVOptions.scala:41) >> at >> org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.buildReader(CSVFileFormat.scala:105) >> at >> org.apache.spark.sql.execution.datasources.FileFormat$class.buildReaderWithPartitionValues(FileFormat.scala:129) >> at >> org.apache.spark.sql.execution.datasources.TextBasedFileFormat.buildReaderWithPartitionValues(FileFormat.scala:165) >> at >> org.apache.spark.sql.execution.FileSourceScanExec.inputRDD$lzycompute(DataSourceScanExec.scala:312) >> at >> org.apache.spark.sql.execution.FileSourceScanExec.inputRDD(DataSourceScanExec.scala:310) >> at >> org.apache.spark.sql.execution.FileSourceScanExec.inputRDDs(DataSourceScanExec.scala:330) >> at >> org.apache.spark.sql.execution.ProjectExec.inputRDDs(basicPhysicalOperators.scala:41) >> at >> org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:615) >> at >> org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131) >> at >> org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127) >> at >> org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155) >> at >> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) >> at >> org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152) >> at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127) >> at >> org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:247) >> at >> org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:339) >> at >> org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38) >> at org.apache.spark.sql.Dataset.org >> $apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3383) >> at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2544) >> at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2544) >> at org.apache.spark.sql.Dataset$$anonfun$53.apply(Dataset.scala:3364) >> at >> org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78) >> at >> org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125) >> at >> org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73) >> at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3363) >> at org.apache.spark.sql.Dataset.head(Dataset.scala:2544) >> at org.apache.spark.sql.Dataset.take(Dataset.scala:2758) >> at org.apache.spark.sql.Dataset.getRows(Dataset.scala:254) >> at org.apache.spark.sql.Dataset.showString(Dataset.scala:291) >> at org.apache.spark.sql.Dataset.show(Dataset.scala:745) >> at org.apache.spark.sql.Dataset.show(Dataset.scala:704) >> at org.apache.spark.sql.Dataset.show(Dataset.scala:713) >> at com.join.ftp.reader.FtpReader.readFrom(FtpReader.scala:40) >> at com.join.synctool$.main(synctool.scala:41) >> at com.join.synctool.main(synctool.scala) >> 21/08/09 11:15:08 INFO SparkContext: Invoking stop() from sh
Is the pandas version in doc of using pyarrow in spark wrong
The doc says that the minimum supported pandas version is 0.23.2 which is only supported in python2. IIRC, python2 is not supported in pyspark a long time ago. Can any one confirm whether the doc is wrong and what is the right version of pandas and pyarrow ? https://spark.apache.org/docs/latest/api/python/user_guide/arrow_pandas.html#recommended-pandas-and-pyarrow-versions -- Best Regards Jeff Zhang
unsubscribe
unsubscribe Regards, Vijay Gharge
Performance of PySpark jobs on the Kubernetes cluster
Hi, I have a basic question to ask. I am running a Google k8s cluster (AKA GKE) with three nodes each having configuration below e2-standard-2 (2 vCPUs, 8 GB memory) spark-submit is launched from another node (actually a data proc single node that I have just upgraded to e2-custom (4 vCPUs, 8 GB mem). We call this the launch node OK I know that the cluster is not much but Google was complaining about the launch node hitting 100% cpus. So I added two more cpus to it. It appears that despite using k8s as the computational cluster, the burden falls upon the launch node! The cpu utilisation for launch node shown below [image: image.png] The dip is when 2 more cpus were added to it so it had to reboot. so around %70 usage The combined cpu usage for GKE nodes is shown below: [image: image.png] Never goes above 20%! I can see that the drive and executors as below: k get pods -n spark NAME READY STATUSRESTARTS AGE pytest-c958c97b2c52b6ed-driver 1/1 Running 0 69s randomdatabigquery-e68a8a7b2c52f468-exec-1 1/1 Running 0 51s randomdatabigquery-e68a8a7b2c52f468-exec-2 1/1 Running 0 51s randomdatabigquery-e68a8a7b2c52f468-exec-3 0/1 Pending 0 51s It is a PySpark 3.1.1 image using java 8 and pushing random data generated into Google BigQuery data warehouse. The last executor (exec-3) seems to be just pending. The spark-submit is as below: spark-submit --verbose \ --properties-file ${property_file} \ --master k8s://https://$KUBERNETES_MASTER_IP:443 \ --deploy-mode cluster \ --name pytest \ --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./pyspark_venv/bin/python \ --py-files $CODE_DIRECTORY/DSBQ.zip \ --conf spark.kubernetes.namespace=$NAMESPACE \ --conf spark.executor.memory=5000m \ --conf spark.network.timeout=300 \ --conf spark.executor.instances=3 \ --conf spark.kubernetes.driver.limit.cores=1 \ --conf spark.driver.cores=1 \ --conf spark.executor.cores=1 \ --conf spark.executor.memory=2000m \ --conf spark.kubernetes.driver.docker.image=${IMAGEGCP} \ --conf spark.kubernetes.executor.docker.image=${IMAGEGCP} \ --conf spark.kubernetes.container.image=${IMAGEGCP} \ --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark-bq \ --conf spark.driver.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true" \ --conf spark.executor.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true" \ --conf spark.sql.execution.arrow.pyspark.enabled="true" \ $CODE_DIRECTORY/${APPLICATION} Aren't the driver and executors running on K8s cluster? So why is the launch node heavily used but k8s cluster is underutilized? Thanks *Disclaimer:* Use it at your own risk. Any and all responsibility for any loss, damage or destruction of data or any other property which may arise from relying on this email's technical content is explicitly disclaimed. The author will in no case be liable for any monetary damages arising from such loss, damage or destruction.
How can I write data to ftp
DF.write.format("csv") .option("timestampFormat", "/MM/dd HH:mm:ss ZZ") .mode(SaveMode.Overwrite) .save("ftp://ftp:ftp@127.0.0.1:21/sparkftp/write/";) I get a error 21/08/10 08:32:04 WARN FileOutputCommitter: Could not delete ftp://ftpuser:ftpuser@10.3.87.51:21/sparkftp/write/_temporary/0/_temporary/attempt_20210810082742__m_00_0 21/08/10 08:32:04 ERROR FileFormatWriter: Job job_20210810082742_ aborted. 21/08/10 08:32:04 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0) org.apache.spark.SparkException: Task failed while writing rows. at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:257) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:170) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:169) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at org.apache.spark.scheduler.Task.run(Task.scala:121) at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.io.IOException: Failed to rename FileStatus{path=ftp://ftpuser:ftpuser@10.3.87.51:21/sparkftp/write/_temporary/0/_temporary/attempt_20210810082742__m_00_0/part-0-e2a4ba4d-18a3-4e97-b7f1-b7a7a65b777e-c000.csv; isDirectory=false; length=39; replication=1; blocksize=4096; modification_time=162852648; access_time=0; owner=1027; group=1027; permission=rw-r--r--; isSymlink=false; hasAcl=false; isEncrypted=false; isErasureCoded=false} to ftp://ftpuser:ftpuser@10.3.87.51:21/sparkftp/write/part-0-e2a4ba4d-18a3-4e97-b7f1-b7a7a65b777e-c000.csv at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.mergePaths(FileOutputCommitter.java:462) at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.mergePaths(FileOutputCommitter.java:475) at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitTask(FileOutputCommitter.java:586) at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitTask(FileOutputCommitter.java:549) at org.apache.spark.mapred.SparkHadoopMapRedUtil$.performCommit$1(SparkHadoopMapRedUtil.scala:50) at org.apache.spark.mapred.SparkHadoopMapRedUtil$.commitTask(SparkHadoopMapRedUtil.scala:77) at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.commitTask(HadoopMapReduceCommitProtocol.scala:225) at org.apache.spark.sql.execution.datasources.FileFormatDataWriter.commit(FileFormatDataWriter.scala:78) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:247) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:242) at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1442) at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:248) ... 10 more 21/08/10 08:32:04 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, localhost, executor driver): org.apache.spark.SparkException: Task failed while writing rows. at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:257) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:170) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:169) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at org.apache.spark.scheduler.Task.run(Task.scala:121) at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.io.IOException: Failed to rename FileStatus{path=ftp://ftpuser:ftpuser@10.3.87.51:21/sparkftp/write/_temporary/0/_temporary/attempt_20210810082742__m_00_0/part-0-e2a4ba4d-18a3-4e97-b7f1-b7a7a65b777e-c000.csv; isDirectory=false; length=39; replicati
Unsubscribe
Unsubscribe