14_spark3_pyspark_shared_variable_acc.ipynb

{"cells":[{"cell_type":"markdown","source":["## Shared variables e accumulators\nNeste notebook iremos aprender a trabalhar com outras APIs low level do Spark.\n\n<p>Você pode usar accumulators para contabilizar dados de todas as tarefas em uma única variável compartilhada.\nPor exemplo, contar quantos registros foram processados em cada tarefa.\n\n<p>Você também pode usar broadcast variables para armazenar variáveis em todos os worker nodes e evitar\nque os mesmos tenham que trocar informações entre eles para ter acesso a este tipo de dado.<br>\nSe for necessário unir dados de um Dataframe grande e um Dataframe pequeno, uma boa forma de otimização é fazer o broadcast do Dataframe pequeno.<br>\n  Veremos este exemplo a seguir"],"metadata":{}},{"cell_type":"markdown","source":["### Importando as bibliotecas\nNesta etapa iremos apenas importar todas as bibliotecas e funções necessárias para rodar o programa"],"metadata":{}},{"cell_type":"code","source":["from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import broadcast\nfrom pyspark.sql.types import StructField, StructType, StringType, LongType, DoubleType, IntegerType"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"]}}],"execution_count":3},{"cell_type":"markdown","source":["### Criando uma SparkSession\nPor meio de uma SparkSession terei acesso ao SparkContext da minha aplicação."],"metadata":{}},{"cell_type":"code","source":["spark = SparkSession.builder.appName('Select').getOrCreate()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"]}}],"execution_count":5},{"cell_type":"markdown","source":["### Criando um Dataframe manualmente\nNesta etapa estamos criando os dados que utilizaremos neste notebook"],"metadata":{}},{"cell_type":"code","source":["# Carrega dados usando schema\nschema = StructType([\n    StructField(\"job_id\", IntegerType(), True),\n    StructField(\"desc\", StringType(), True)\n])\n\n# Cria as linhas do nosso futuro dataframe\nnewRows = [\n    [30, \"Cientista de dados\"],\n    [20, \"Dev Java\"],\n    [10, None]\n]\n\n# Cria um RDD de Rows\nparallelizedRows = spark.sparkContext.parallelize(newRows)\n\n# Cria um dataframe a partir do RDD que criamos anteriormente\ndados_manual = spark.createDataFrame(parallelizedRows, schema)\n\n# Mostra as informações do dataframe\ndados_manual.show()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+------+------------------+\njob_id|              desc|\n+------+------------------+\n    30|Cientista de dados|\n    20|          Dev Java|\n    10|              null|\n+------+------------------+\n\n</div>"]}}],"execution_count":7},{"cell_type":"markdown","source":["### Criando um segundo Dataframe manualmente"],"metadata":{}},{"cell_type":"code","source":["# Carrega dados usando schema\nschema = StructType([\n    StructField(\"job_id\", IntegerType(), True),\n    StructField(\"salary\", DoubleType(), True)\n])\n\n# Cria as linhas do nosso futuro dataframe\nnewRows = [\n    [30, 10000.0],\n    [20, 9000.0],\n    [1, 15000.0],\n    [2, 2000.0],\n    [3, 3000.0]\n]\n\n# Cria um RDD de Rows\nparallelizedRows = spark.sparkContext.parallelize(newRows)\n\n# Cria um dataframe a partir do RDD que criamos anteriormente\nsalarios = spark.createDataFrame(parallelizedRows, schema)\n\n# Mostra as informações do dataframe\nsalarios.show()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">+------+-------+\njob_id| salary|\n+------+-------+\n    30|10000.0|\n    20| 9000.0|\n     1|15000.0|\n     2| 2000.0|\n     3| 3000.0|\n+------+-------+\n\n</div>"]}}],"execution_count":9},{"cell_type":"code","source":["dados_join = dados_manual.join(salarios, dados_manual.job_id == salarios.job_id, \"inner\")"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"]}}],"execution_count":10},{"cell_type":"code","source":["dados_join.explain()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">== Physical Plan ==\n*(3) SortMergeJoin [job_id#1554], [job_id#1567], Inner\n:- Sort [job_id#1554 ASC NULLS FIRST], false, 0\n:  +- Exchange hashpartitioning(job_id#1554, 200), true, [id=#432]\n:     +- *(1) Filter isnotnull(job_id#1554)\n:        +- *(1) Scan ExistingRDD[job_id#1554,desc#1555]\n+- Sort [job_id#1567 ASC NULLS FIRST], false, 0\n   +- Exchange hashpartitioning(job_id#1567, 200), true, [id=#436]\n      +- *(2) Filter isnotnull(job_id#1567)\n         +- *(2) Scan ExistingRDD[job_id#1567,salary#1568]\n\n\n</div>"]}}],"execution_count":11},{"cell_type":"code","source":["salarios.persist()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">Out[10]: DataFrame[job_id: int, salary: double]</div>"]}}],"execution_count":12},{"cell_type":"code","source":["broadcast(salarios)\nsalarios.count()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">Out[29]: 5</div>"]}}],"execution_count":13},{"cell_type":"code","source":["dados_join = dados_manual.join(salarios, dados_manual.job_id == salarios.job_id, \"inner\")\ndados_join.explain()"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">== Physical Plan ==\n*(2) BroadcastHashJoin [job_id#1554], [job_id#1567], Inner, BuildRight\n:- *(2) Filter isnotnull(job_id#1554)\n:  +- *(2) Scan ExistingRDD[job_id#1554,desc#1555]\n+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint))), [id=#583]\n   +- *(1) Filter isnotnull(job_id#1567)\n      +- *(1) ColumnarToRow\n         +- InMemoryTableScan [job_id#1567, salary#1568], [isnotnull(job_id#1567)]\n               +- InMemoryRelation [job_id#1567, salary#1568], StorageLevel(disk, memory, 1 replicas)\n                     +- *(1) Scan ExistingRDD[job_id#1567,salary#1568]\n\n\n</div>"]}}],"execution_count":14},{"cell_type":"markdown","source":["### Broacast de outros objetos"],"metadata":{}},{"cell_type":"code","source":["filiais = [\"SP\", \"BH\", \"RJ\"]\nfiliais_bc = spark.sparkContext.broadcast(filiais)\nfiliais_bc.value"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">Out[13]: [&#39;SP&#39;, &#39;BH&#39;, &#39;RJ&#39;]</div>"]}}],"execution_count":16},{"cell_type":"markdown","source":["### Accumulators"],"metadata":{}},{"cell_type":"code","source":["# Definição do acumulador\ncontador = spark.sparkContext.accumulator(0)\n\ndef contadorFunc(salario):\n    if (salario >= 10000):\n        contador.add(1)\n\n# Chamando sua função diretamente, apenas para testar o acumulador\ncontador.value\ncontadorFunc(10000)\ncontador.value\n\n# Chamando sua função dentro de um dataframe\nsalarios.foreach(lambda row : contadorFunc(row.salary))\ncontador.value"],"metadata":{},"outputs":[{"metadata":{},"output_type":"display_data","data":{"text/html":["<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\">Out[25]: 3</div>"]}}],"execution_count":18},{"cell_type":"markdown","source":["### Obrigado!\nQuer construir uma carreira em Data Science? Acesse meu blog pessoal em https://www.hackinganalytics.com/"],"metadata":{}}],"metadata":{"name":"14_spark3_pyspark_shared_variable_acc","notebookId":2586057201697934},"nbformat":4,"nbformat_minor":0}