Format: application/octet-stream
1 {
2 "cells": [
3 {
4 "cell_type": "code",
5 "execution_count": 1,
6 "id": "d35cbf81-9ec9-45f7-b034-fa296d429f54",
7 "metadata": {},
8 "outputs": [],
9 "source": [
10 "from credentials import *\n",
11 "from dataextractor import *\n",
12 "from pyspark.sql import SparkSession\n",
13 "from pyspark.sql.functions import *\n",
14 "import datetime"
15 ]
16 },
17 {
18 "cell_type": "code",
19 "execution_count": 2,
20 "id": "56454870-c431-4fe0-88dd-26fdbf3821e8",
21 "metadata": {},
22 "outputs": [],
23 "source": [
24 "from pyspark import SparkContext"
25 ]
26 },
27 {
28 "cell_type": "code",
29 "execution_count": 3,
30 "id": "4722d484-7148-4efe-8711-0bd44027ff94",
31 "metadata": {},
32 "outputs": [],
33 "source": [
34 "s= datetime.datetime.now()"
35 ]
36 },
37 {
38 "cell_type": "code",
39 "execution_count": 4,
40 "id": "58e514fe-b19f-485d-a368-98da34f4f021",
41 "metadata": {},
42 "outputs": [],
43 "source": [
44 "# from pyspark import SparkConf\n",
45 "# conf = SparkConf().set(\"spark.hadoop.fs.s3a.access.key\", \"AKIATBUD4GB6PUTB2TOB\") \\\n",
46 "# .set(\"spark.hadoop.fs.s3a.secret.key\", \"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\") \\\n",
47 "# .set('packages', 'net.snowflake:snowflake-jdbc:3.13.26,net.snowflake:spark-snowflake_2.13:2.11.1-spark_3.3')"
48 ]
49 },
50 {
51 "cell_type": "code",
52 "execution_count": 5,
53 "id": "0762f644-5aad-4c86-8c4f-94ae30e8c92b",
54 "metadata": {},
55 "outputs": [],
56 "source": [
57 "# #spark.stop()\n",
58 "# common ='file:///c:/Users/PC/Downloads/'\n",
59 "path1='snowflake-jdbc-3.13.26.jar'\n",
60 "path2='spark-snowflake_2.13-2.11.1-spark_3.3.jar'\n",
61 "p=path1+','+path2\n",
62 "spark = SparkSession.builder \\\n",
63 " .config('spark.jars', p) \\\n",
64 " .appName('dev') \\\n",
65 " .getOrCreate()"
66 ]
67 },
68 {
69 "cell_type": "code",
70 "execution_count": 6,
71 "id": "b9337f3d-3553-4140-9171-93cef9ad191b",
72 "metadata": {},
73 "outputs": [],
74 "source": [
75 "# #spark.stop()\n",
76 "# spark = SparkSession.builder \\\n",
77 "# .config('packages', 'net.snowflake:snowflake-jdbc:3.13.26,net.snowflake:spark-snowflake_2.13:2.11.1-spark_3.3') \\\n",
78 "# .config(\"spark.hadoop.fs.s3a.access.key\", \"AKIATBUD4GB6PUTB2TOB\")\\\n",
79 "# .config(\"spark.hadoop.fs.s3a.secret.key\", \"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\")\\\n",
80 "# .appName('dev') \\\n",
81 "# .getOrCreate()"
82 ]
83 },
84 {
85 "cell_type": "code",
86 "execution_count": 7,
87 "id": "b7dff7ae-d957-4183-9808-1618f4cc4203",
88 "metadata": {},
89 "outputs": [],
90 "source": [
91 "cred_manager= CredentialManager()\n",
92 "dataextractor= DataExtractor(spark, CredentialManager())"
93 ]
94 },
95 {
96 "cell_type": "code",
97 "execution_count": 8,
98 "id": "951b1951-a38e-48c5-9a11-b95b8015ba75",
99 "metadata": {},
100 "outputs": [],
101 "source": [
102 "options={'table':'CUSTOMER_CATALOG',\n",
103 " 'warehouse':'COMPUTE', 'schema':'BUNGEE', 'database':'PETCO_BUNGEE_SHARE_INT'\n",
104 " }"
105 ]
106 },
107 {
108 "cell_type": "code",
109 "execution_count": 9,
110 "id": "6efa9696-85a1-4680-b69f-57f7a0fee010",
111 "metadata": {},
112 "outputs": [],
113 "source": [
114 "df= dataextractor.extract_data('petco-snowflake', options , 'cast(product_id as double)%2 = 0')"
115 ]
116 },
117 {
118 "cell_type": "code",
119 "execution_count": 10,
120 "id": "cd909b6c-a1d1-4c0f-8a69-428d0deaa100",
121 "metadata": {},
122 "outputs": [
123 {
124 "name": "stdout",
125 "output_type": "stream",
126 "text": [
127 "root\n",
128 " |-- DATETIME: timestamp (nullable = false)\n",
129 " |-- PRODUCT_ID: decimal(38,0) (nullable = false)\n",
130 " |-- PARENT_PRODUCT_PART_DOUBLE: string (nullable = true)\n",
131 " |-- UPC: string (nullable = true)\n",
132 " |-- PRODUCT_NAME: string (nullable = true)\n",
133 " |-- PRODUCT_DESCRIPTION: string (nullable = true)\n",
134 " |-- PRODUCT_SIZE: string (nullable = true)\n",
135 " |-- PRODUCT_COLOR: string (nullable = true)\n",
136 " |-- PRODUCT_PURCHASE_BRAND: string (nullable = true)\n",
137 " |-- PRIMARY_VENDOR_NAME: string (nullable = true)\n",
138 " |-- PRODUCT_CATEGORY_LEVEL1: string (nullable = true)\n",
139 " |-- PRODUCT_CATEGORY_LEVEL2: string (nullable = true)\n",
140 " |-- PRODUCT_CATEGORY_LEVEL3: string (nullable = true)\n",
141 " |-- PRODUCT_CATEGORY_LEVEL4: string (nullable = true)\n",
142 " |-- PRODUCT_CATEGORY_LEVEL5: string (nullable = true)\n",
143 " |-- PRODUCT_CATEGORY_LEVEL6: string (nullable = true)\n",
144 " |-- PRODUCT_FULLIMAGE: string (nullable = true)\n",
145 " |-- PRODUCT_URL: string (nullable = true)\n",
146 " |-- PRODUCT_DISCONTINUED_FLAG: date (nullable = true)\n",
147 " |-- PRODUCT_PUBLISHED_FLAG: decimal(38,0) (nullable = true)\n",
148 " |-- PRODUCT_WEIGHT: string (nullable = true)\n",
149 " |-- PRODUCT_WEIGHT_UOM: string (nullable = true)\n",
150 " |-- PRODUCT_LENGTH: decimal(10,2) (nullable = true)\n",
151 " |-- PRODUCT_WIDTH: decimal(10,2) (nullable = true)\n",
152 " |-- PRODUCT_HEIGHT: decimal(10,2) (nullable = true)\n",
153 "\n"
154 ]
155 }
156 ],
157 "source": [
158 "df.printSchema()"
159 ]
160 },
161 {
162 "cell_type": "code",
163 "execution_count": 11,
164 "id": "cb044fbf-5e1a-4358-a305-f52c29b0ce2d",
165 "metadata": {},
166 "outputs": [],
167 "source": [
168 "grouped= df.groupby('PRODUCT_COLOR').agg(count('PRODUCT_ID').alias('count') )"
169 ]
170 },
171 {
172 "cell_type": "code",
173 "execution_count": 12,
174 "id": "ce22da84-af22-4320-addf-2c898fe887b1",
175 "metadata": {},
176 "outputs": [
177 {
178 "name": "stdout",
179 "output_type": "stream",
180 "text": [
181 "+----------------+------+\n",
182 "| PRODUCT_COLOR| count|\n",
183 "+----------------+------+\n",
184 "| null|109251|\n",
185 "|Other or Unknown| 6179|\n",
186 "| Multi| 6052|\n",
187 "| Blue| 3602|\n",
188 "| Pink| 2816|\n",
189 "| Red| 2634|\n",
190 "| Black| 2599|\n",
191 "| Gray| 1967|\n",
192 "| Brown| 1742|\n",
193 "| Green| 1662|\n",
194 "| White or Ivory| 1071|\n",
195 "| Orange| 601|\n",
196 "| Purple| 571|\n",
197 "| Yellow| 475|\n",
198 "| Metallic| 338|\n",
199 "| Transparent| 67|\n",
200 "+----------------+------+\n",
201 "\n"
202 ]
203 }
204 ],
205 "source": [
206 "grouped.repartition(1).orderBy(grouped['count'].desc()).show()"
207 ]
208 },
209 {
210 "cell_type": "code",
211 "execution_count": 13,
212 "id": "9972dabe-84d8-49e0-9693-a2f5d35b9f8e",
213 "metadata": {},
214 "outputs": [
215 {
216 "name": "stdout",
217 "output_type": "stream",
218 "text": [
219 "+----------------------+\n",
220 "|PRODUCT_PUBLISHED_FLAG|\n",
221 "+----------------------+\n",
222 "| 1|\n",
223 "| null|\n",
224 "| 0|\n",
225 "+----------------------+\n",
226 "\n"
227 ]
228 }
229 ],
230 "source": [
231 "df.select('PRODUCT_PUBLISHED_FLAG').distinct().show()"
232 ]
233 },
234 {
235 "cell_type": "code",
236 "execution_count": 14,
237 "id": "60fcc060-70ee-4dc0-b112-7f5d0c65419d",
238 "metadata": {},
239 "outputs": [],
240 "source": [
241 "df =df.withColumn('is_discontinued' , when(isnull(col('PRODUCT_DISCONTINUED_FLAG')), True).otherwise(False) )"
242 ]
243 },
244 {
245 "cell_type": "code",
246 "execution_count": 15,
247 "id": "b66dce0c-8a2f-4e59-b329-54c86c238c80",
248 "metadata": {},
249 "outputs": [],
250 "source": [
251 "df =df.withColumn('is_active' , when(isnull(col('PRODUCT_DISCONTINUED_FLAG')), 0).otherwise(1) )"
252 ]
253 },
254 {
255 "cell_type": "code",
256 "execution_count": 16,
257 "id": "5396ff66-7b0f-4568-82b5-0c17f4a99e29",
258 "metadata": {},
259 "outputs": [
260 {
261 "name": "stdout",
262 "output_type": "stream",
263 "text": [
264 "+---------+---------------+-----------------+\n",
265 "|is_active|is_discontinued|count(PRODUCT_ID)|\n",
266 "+---------+---------------+-----------------+\n",
267 "| 0| true| 35090|\n",
268 "| 1| false| 106537|\n",
269 "+---------+---------------+-----------------+\n",
270 "\n"
271 ]
272 }
273 ],
274 "source": [
275 "df.groupby('is_active', 'is_discontinued').agg(count(col('PRODUCT_ID'))).show()"
276 ]
277 },
278 {
279 "cell_type": "code",
280 "execution_count": 17,
281 "id": "91e016f5-26ff-49ec-92cc-cf85a7a08d43",
282 "metadata": {},
283 "outputs": [],
284 "source": [
285 "grouped=df.filter(col('is_active')==1).groupby('PRODUCT_PURCHASE_BRAND').agg(count(col('PRODUCT_ID')).alias('count'))"
286 ]
287 },
288 {
289 "cell_type": "code",
290 "execution_count": 18,
291 "id": "898ea84a-cdff-457c-b46c-c68d725c0b8d",
292 "metadata": {},
293 "outputs": [
294 {
295 "name": "stdout",
296 "output_type": "stream",
297 "text": [
298 "+----------------------+-----+\n",
299 "|PRODUCT_PURCHASE_BRAND|count|\n",
300 "+----------------------+-----+\n",
301 "| UNASSIGNED|25060|\n",
302 "| PETCO BRAND| 6303|\n",
303 "| SIMPLY SHE (CUSTO...| 3313|\n",
304 "| DICKENS CLOSET| 2947|\n",
305 "| BOND & CO.| 1844|\n",
306 "+----------------------+-----+\n",
307 "only showing top 5 rows\n",
308 "\n"
309 ]
310 }
311 ],
312 "source": [
313 "grouped.repartition(1).orderBy(grouped['count'].desc()).show(5)"
314 ]
315 },
316 {
317 "cell_type": "code",
318 "execution_count": 19,
319 "id": "b8530fd2-7cef-4928-bde9-54d22135a87e",
320 "metadata": {},
321 "outputs": [
322 {
323 "data": {
324 "text/plain": [
325 "1830"
326 ]
327 },
328 "execution_count": 19,
329 "metadata": {},
330 "output_type": "execute_result"
331 }
332 ],
333 "source": [
334 "grouped.count()"
335 ]
336 },
337 {
338 "cell_type": "code",
339 "execution_count": 20,
340 "id": "3a5be4c2-9a8b-4c2d-986c-85705284294a",
341 "metadata": {},
342 "outputs": [],
343 "source": [
344 "grouped2=df.filter(col('is_active')==1).groupby('PRODUCT_PURCHASE_BRAND')\\\n",
345 " .agg(collect_set(substring(col('UPC'),1,5)).alias('upc5'),\n",
346 " count_distinct( substring(col('UPC'),1,5)).alias('count_upc5')\n",
347 " )"
348 ]
349 },
350 {
351 "cell_type": "code",
352 "execution_count": 21,
353 "id": "697a1c98-0d40-4436-af61-7de33dd90453",
354 "metadata": {},
355 "outputs": [
356 {
357 "data": {
358 "text/plain": [
359 "1830"
360 ]
361 },
362 "execution_count": 21,
363 "metadata": {},
364 "output_type": "execute_result"
365 }
366 ],
367 "source": [
368 "grouped2.count()"
369 ]
370 },
371 {
372 "cell_type": "code",
373 "execution_count": 22,
374 "id": "1e8c028c-ad64-475f-9c93-17a1889f0626",
375 "metadata": {},
376 "outputs": [],
377 "source": [
378 "final_df= grouped.repartition(1).join(grouped2.repartition(1),['PRODUCT_PURCHASE_BRAND'], 'inner')"
379 ]
380 },
381 {
382 "cell_type": "code",
383 "execution_count": 23,
384 "id": "83d9dc50-131d-4ead-a1df-bde717bd266c",
385 "metadata": {},
386 "outputs": [
387 {
388 "name": "stdout",
389 "output_type": "stream",
390 "text": [
391 "+----------------------+-----+--------------------+----------+\n",
392 "|PRODUCT_PURCHASE_BRAND|count| upc5|count_upc5|\n",
393 "+----------------------+-----+--------------------+----------+\n",
394 "| 2 FOR $5| 114|[80044, 74677, 61...| 4|\n",
395 "| 2 FOR 5| 2| [74677]| 1|\n",
396 "| 26 BARS AND A BAND| 57| [71812, 70510]| 2|\n",
397 "| 3 DOG PET SUPPLY| 1| [85975]| 1|\n",
398 "| 3M MASKING & PACK...| 39|[50051, 02120, 05...| 5|\n",
399 "+----------------------+-----+--------------------+----------+\n",
400 "only showing top 5 rows\n",
401 "\n"
402 ]
403 }
404 ],
405 "source": [
406 "final_df.show(5)"
407 ]
408 },
409 {
410 "cell_type": "code",
411 "execution_count": null,
412 "id": "9d4c489b-d2c8-4542-ad36-4a8410383908",
413 "metadata": {},
414 "outputs": [],
415 "source": []
416 },
417 {
418 "cell_type": "code",
419 "execution_count": 24,
420 "id": "779a43d6-b977-4152-861d-da280edbec26",
421 "metadata": {},
422 "outputs": [
423 {
424 "name": "stdout",
425 "output_type": "stream",
426 "text": [
427 "0:01:18.503431\n"
428 ]
429 }
430 ],
431 "source": [
432 "print(datetime.datetime.now() -s )"
433 ]
434 },
435 {
436 "cell_type": "code",
437 "execution_count": 25,
438 "id": "63a0c9ac-7728-4976-82db-912d369b818a",
439 "metadata": {},
440 "outputs": [],
441 "source": [
442 "s=datetime.datetime.now()"
443 ]
444 },
445 {
446 "cell_type": "code",
447 "execution_count": 31,
448 "id": "10555399-57e1-4de7-9841-16ab29604526",
449 "metadata": {},
450 "outputs": [],
451 "source": [
452 "sc = spark.sparkContext\n",
453 "access_key='AKIATBUD4GB6PUTB2TOB'\n",
454 "secret_key='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'\n",
455 "# remove this block if use core-site.xml and env variable\n",
456 "sc._jsc.hadoopConfiguration().set(\"fs.s3.awsAccessKeyId\", access_key)\n",
457 "sc._jsc.hadoopConfiguration().set(\"fs.s3n.awsAccessKeyId\", access_key)\n",
458 "sc._jsc.hadoopConfiguration().set(\"fs.s3a.access.key\", access_key)\n",
459 "sc._jsc.hadoopConfiguration().set(\"fs.s3.awsSecretAccessKey\", secret_key)\n",
460 "sc._jsc.hadoopConfiguration().set(\"fs.s3n.awsSecretAccessKey\", secret_key)\n",
461 "sc._jsc.hadoopConfiguration().set(\"fs.s3a.secret.key\", secret_key)\n",
462 "sc._jsc.hadoopConfiguration().set(\"fs.s3n.impl\", \"org.apache.hadoop.fs.s3native.NativeS3FileSystem\")\n",
463 "sc._jsc.hadoopConfiguration().set(\"fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
464 "sc._jsc.hadoopConfiguration().set(\"fs.s3.impl\", \"org.apache.hadoop.fs.s3.S3FileSystem\")"
465 ]
466 },
467 {
468 "cell_type": "code",
469 "execution_count": 34,
470 "id": "6b43c7de-38b7-4e92-b6fe-48ebe4a2c483",
471 "metadata": {},
472 "outputs": [
473 {
474 "ename": "Py4JJavaError",
475 "evalue": "An error occurred while calling o160.parquet.\n: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme \"s3fs\"\r\n\tat org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3443)\r\n\tat org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3466)\r\n\tat org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174)\r\n\tat org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3574)\r\n\tat org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3521)\r\n\tat org.apache.hadoop.fs.FileSystem.get(FileSystem.java:540)\r\n\tat org.apache.hadoop.fs.Path.getFileSystem(Path.java:365)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.planForWritingFileFormat(DataSource.scala:461)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.planForWriting(DataSource.scala:558)\r\n\tat org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:390)\r\n\tat org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:363)\r\n\tat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:239)\r\n\tat org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:793)\r\n\tat java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)\r\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:578)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\r\n\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\r\n\tat java.base/java.lang.Thread.run(Thread.java:1589)\r\n",
476 "output_type": "error",
477 "traceback": [
478 "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
479 "\u001b[1;31mPy4JJavaError\u001b[0m Traceback (most recent call last)",
480 "Cell \u001b[1;32mIn[34], line 1\u001b[0m\n\u001b[1;32m----> 1\u001b[0m \u001b[43mfinal_df\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mwrite\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mparquet\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[38;5;124;43ms3fs://shubham-test-bucket/testing/data/petco\u001b[39;49m\u001b[38;5;124;43m'\u001b[39;49m\u001b[43m)\u001b[49m\n",
481 "File \u001b[1;32m~\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\site-packages\\pyspark\\sql\\readwriter.py:1140\u001b[0m, in \u001b[0;36mDataFrameWriter.parquet\u001b[1;34m(self, path, mode, partitionBy, compression)\u001b[0m\n\u001b[0;32m 1138\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mpartitionBy(partitionBy)\n\u001b[0;32m 1139\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_set_opts(compression\u001b[38;5;241m=\u001b[39mcompression)\n\u001b[1;32m-> 1140\u001b[0m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_jwrite\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mparquet\u001b[49m\u001b[43m(\u001b[49m\u001b[43mpath\u001b[49m\u001b[43m)\u001b[49m\n",
482 "File \u001b[1;32m~\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\site-packages\\py4j\\java_gateway.py:1321\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m 1315\u001b[0m command \u001b[38;5;241m=\u001b[39m proto\u001b[38;5;241m.\u001b[39mCALL_COMMAND_NAME \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m 1316\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mcommand_header \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m 1317\u001b[0m args_command \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m 1318\u001b[0m proto\u001b[38;5;241m.\u001b[39mEND_COMMAND_PART\n\u001b[0;32m 1320\u001b[0m answer \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client\u001b[38;5;241m.\u001b[39msend_command(command)\n\u001b[1;32m-> 1321\u001b[0m return_value \u001b[38;5;241m=\u001b[39m \u001b[43mget_return_value\u001b[49m\u001b[43m(\u001b[49m\n\u001b[0;32m 1322\u001b[0m \u001b[43m \u001b[49m\u001b[43manswer\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mgateway_client\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mtarget_id\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mname\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m 1324\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m temp_arg \u001b[38;5;129;01min\u001b[39;00m temp_args:\n\u001b[0;32m 1325\u001b[0m temp_arg\u001b[38;5;241m.\u001b[39m_detach()\n",
483 "File \u001b[1;32m~\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\site-packages\\pyspark\\sql\\utils.py:190\u001b[0m, in \u001b[0;36mcapture_sql_exception.<locals>.deco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m 188\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mdeco\u001b[39m(\u001b[38;5;241m*\u001b[39ma: Any, \u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mkw: Any) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m Any:\n\u001b[0;32m 189\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m--> 190\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43mf\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[43ma\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[43mkw\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m 191\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m Py4JJavaError \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[0;32m 192\u001b[0m converted \u001b[38;5;241m=\u001b[39m convert_exception(e\u001b[38;5;241m.\u001b[39mjava_exception)\n",
484 "File \u001b[1;32m~\\AppData\\Local\\Programs\\Python\\Python311\\Lib\\site-packages\\py4j\\protocol.py:326\u001b[0m, in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m 324\u001b[0m value \u001b[38;5;241m=\u001b[39m OUTPUT_CONVERTER[\u001b[38;5;28mtype\u001b[39m](answer[\u001b[38;5;241m2\u001b[39m:], gateway_client)\n\u001b[0;32m 325\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m answer[\u001b[38;5;241m1\u001b[39m] \u001b[38;5;241m==\u001b[39m REFERENCE_TYPE:\n\u001b[1;32m--> 326\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m Py4JJavaError(\n\u001b[0;32m 327\u001b[0m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mAn error occurred while calling \u001b[39m\u001b[38;5;132;01m{0}\u001b[39;00m\u001b[38;5;132;01m{1}\u001b[39;00m\u001b[38;5;132;01m{2}\u001b[39;00m\u001b[38;5;124m.\u001b[39m\u001b[38;5;130;01m\\n\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39m\n\u001b[0;32m 328\u001b[0m \u001b[38;5;28mformat\u001b[39m(target_id, \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m.\u001b[39m\u001b[38;5;124m\"\u001b[39m, name), value)\n\u001b[0;32m 329\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m 330\u001b[0m \u001b[38;5;28;01mraise\u001b[39;00m Py4JError(\n\u001b[0;32m 331\u001b[0m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mAn error occurred while calling \u001b[39m\u001b[38;5;132;01m{0}\u001b[39;00m\u001b[38;5;132;01m{1}\u001b[39;00m\u001b[38;5;132;01m{2}\u001b[39;00m\u001b[38;5;124m. Trace:\u001b[39m\u001b[38;5;130;01m\\n\u001b[39;00m\u001b[38;5;132;01m{3}\u001b[39;00m\u001b[38;5;130;01m\\n\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39m\n\u001b[0;32m 332\u001b[0m \u001b[38;5;28mformat\u001b[39m(target_id, \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m.\u001b[39m\u001b[38;5;124m\"\u001b[39m, name, value))\n",
485 "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o160.parquet.\n: org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme \"s3fs\"\r\n\tat org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:3443)\r\n\tat org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3466)\r\n\tat org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174)\r\n\tat org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3574)\r\n\tat org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3521)\r\n\tat org.apache.hadoop.fs.FileSystem.get(FileSystem.java:540)\r\n\tat org.apache.hadoop.fs.Path.getFileSystem(Path.java:365)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.planForWritingFileFormat(DataSource.scala:461)\r\n\tat org.apache.spark.sql.execution.datasources.DataSource.planForWriting(DataSource.scala:558)\r\n\tat org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:390)\r\n\tat org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:363)\r\n\tat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:239)\r\n\tat org.apache.spark.sql.DataFrameWriter.parquet(DataFrameWriter.scala:793)\r\n\tat java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)\r\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:578)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)\r\n\tat py4j.ClientServerConnection.run(ClientServerConnection.java:106)\r\n\tat java.base/java.lang.Thread.run(Thread.java:1589)\r\n"
486 ]
487 }
488 ],
489 "source": [
490 "final_df.write.parquet('s3fs://shubham-test-bucket/testing/data/petco')"
491 ]
492 },
493 {
494 "cell_type": "code",
495 "execution_count": null,
496 "id": "259cae68-d414-46a8-b637-de070dbe0eb1",
497 "metadata": {},
498 "outputs": [],
499 "source": [
500 "print(datetime.datetime.now() -s )"
501 ]
502 },
503 {
504 "cell_type": "code",
505 "execution_count": null,
506 "id": "12b937d9-b40d-414c-ab8b-5404f44ad68b",
507 "metadata": {},
508 "outputs": [],
509 "source": []
510 }
511 ],
512 "metadata": {
513 "kernelspec": {
514 "display_name": "Python 3 (ipykernel)",
515 "language": "python",
516 "name": "python3"
517 },
518 "language_info": {
519 "codemirror_mode": {
520 "name": "ipython",
521 "version": 3
522 },
523 "file_extension": ".py",
524 "mimetype": "text/x-python",
525 "name": "python",
526 "nbconvert_exporter": "python",
527 "pygments_lexer": "ipython3",
528 "version": "3.11.1"
529 }
530 },
531 "nbformat": 4,
532 "nbformat_minor": 5
533 }
534