Data-Warehouse-AWS-Pipeline-Chat-API.ipynb

{"cells":[{"cell_type":"markdown","metadata":{},"source":["# Pipeline e Análise de Dados de Mensagens do Telegram\n","\n","[Rafael Barbosa](https://www.linkedin.com/in/barbosa89/)\n","<br>\n","[Marcelo Barbosa](https://www.linkedin.com/in/marcelogomesbarbosa)\n","<br>\n","[André Perez](https://www.linkedin.com/in/andremarcosperez/)\n","<br>\n","[Mariane Neiva](https://www.linkedin.com/in/mariane-neiva/)\n","\n","\n","---\n","\n","A seguir um modelo de pipeline com guia de configuração utilizando Amazon Web Services, Python e SQL para ingerir, extrair, transformar e carregar dados de mensagens de um chat do Telegram, seguindo o modelo da imagem a seguir.\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/Telegram%20Data%20Pipeline%20Schema.png?raw=true)\n","\n","Insights importantes são inferidos a partir dos dados de mensagens de chats, identificando gargalos mais comuns e áreas para melhoria, guiando o desempenho de agentes, promovendo ofertas personalizadas, gerando leads de forma eficiente, identificando tendências e entendendo necessidades dos clientes, avaliando a satisfação do cliente, e claro melhorando a comunicação e colaboração entre equipes.\n","\n","Este passo a passo processa dados transacionais (o que acontece), em dados analíticos (por que e como acontece) e deve ter seus parâmetros configurados."]},{"cell_type":"markdown","metadata":{},"source":["\n","## 0/3\\. Fonte de Dados\n","\n","Dados de mensagens captadas por um [**Telegram bot**](https://core.telegram.org/bots/api) via `API token` disponibilizado no `BotFather` do Telegram. [Crie e configure um Telegram bot API token aqui.](https://github.com/rafie-b/Profession-Data-Analyst/blob/main/Pipeline-Data-API-Telegram-bot-Profissao_Analista_de_dados_M42_Exercicio_pt1_Rafael_Barbosa.ipynb) Guarde o \"token\" em \"base_url\" usando a célula de código Python abaixo."]},{"cell_type":"code","execution_count":1,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:31:47.433255Z","iopub.status.busy":"2024-06-09T23:31:47.432841Z","iopub.status.idle":"2024-06-09T23:31:50.900257Z","shell.execute_reply":"2024-06-09T23:31:50.899148Z","shell.execute_reply.started":"2024-06-09T23:31:47.433220Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["insert Telegram bot API Token\n"]},{"name":"stdout","output_type":"stream","text":[" ··············································\n"]},{"name":"stdout","output_type":"stream","text":["base_url stored\n"]}],"source":["### Salva o API token em url_base\n","# A `url` base é comum a todos os métodos da API\n","from getpass import getpass\n","import requests\n","\n","print(\"insert Telegram bot API Token\")\n","token = getpass()\n","\n","base_url = f'https://api.telegram.org/bot{token}'\n","print(\"base_url stored\")"]},{"cell_type":"code","execution_count":6,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:36:15.965983Z","iopub.status.busy":"2024-06-09T23:36:15.965605Z","iopub.status.idle":"2024-06-09T23:36:16.708108Z","shell.execute_reply":"2024-06-09T23:36:16.706874Z","shell.execute_reply.started":"2024-06-09T23:36:15.965955Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["{\n","  \"ok\": true,\n","  \"result\": {\n","    \"id\": 7084060512,\n","    \"is_bot\": true,\n","    \"first_name\": \"ebac_m42_bot\",\n","    \"username\": \"ebacm42_bot\",\n","    \"can_join_groups\": false,\n","    \"can_read_all_group_messages\": false,\n","    \"supports_inline_queries\": false,\n","    \"can_connect_to_business\": false\n","  }\n","}\n"]}],"source":["### `getMe` retorna informações sobre o bot\n","import json\n","import requests\n","\n","response = requests.get(url=f'{base_url}/getMe')\n","print(json.dumps(json.loads(response.text), indent=2))"]},{"cell_type":"code","execution_count":8,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:36:39.790433Z","iopub.status.busy":"2024-06-09T23:36:39.789999Z","iopub.status.idle":"2024-06-09T23:36:40.549091Z","shell.execute_reply":"2024-06-09T23:36:40.547879Z","shell.execute_reply.started":"2024-06-09T23:36:39.790399Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["{\n","  \"ok\": true,\n","  \"result\": [\n","    {\n","      \"update_id\": 827865390,\n","      \"message\": {\n","        \"message_id\": 68,\n","        \"from\": {\n","          \"id\": 7162819601,\n","          \"is_bot\": false,\n","          \"first_name\": \"Rafie\",\n","          \"language_code\": \"en\"\n","        },\n","        \"chat\": {\n","          \"id\": -1001993072220,\n","          \"title\": \"EBAC M42 BOT\",\n","          \"type\": \"supergroup\"\n","        },\n","        \"date\": 1717976196,\n","        \"text\": \"up up\"\n","      }\n","    }\n","  ]\n","}\n"]}],"source":["### `getUpdates` revela a estrutura de dados de mensagens captadas pelo bot\n","import json\n","import requests\n","\n","response = requests.get(url=f'{base_url}/getUpdates')\n","print(json.dumps(json.loads(response.text), indent=2))"]},{"cell_type":"markdown","metadata":{},"source":["0.1. Guarde o número encontrado no 'id' do item 'chat' do `getUpdates` para configurar a variável de ambiente do `AWS Lambda`."]},{"cell_type":"markdown","metadata":{},"source":["# 1/3. Ingestão\n","\n","**1.1.** `AWS S3 bucket` com sufixo `-raw` para **dados crus**;\n","\n","**1.2.** `AWS Lambda` **função Python** que recebe e armazena dados crus no formato JSON, no `AWS S3 bucket` sufixo `-raw`;\n","\n","1.2.1. variáveis de ambiente para **nome do bucket de dados crus** (AWS_S3_BUCKET) e para **id do chat do Telegram** (TELEGRAM_CHAT_ID) em \"Configuração > Variáveis de Ambiente > Editar\" no painel da função;\n","\n","1.2.2. permissão para escrita no bucket de dados crus em \"Configuração > Permissôes > Nome da Função [<Função>]\" no painel da função e no `AWS IAM` \"Adicionar permissões > Anexar políticas\" para o `AWS S3 bucket` sufixo `-raw`;\n","\n","\n","```python\n","### código Python da função `AWS Lambda` criada para dados crus:\n","\n","# persistir as mensagens captadas pelo bot do Telegram em um bucket do AWS S3\n","# Recebe a mensagem no parâmetro event;\n","# Verifica se a mensagem tem origem no grupo do Telegram correto;\n","# Persiste a mensagem no formato JSON no bucket do AWS S3;\n","# Retorna uma mensagem de sucesso (código de retorno HTTP igual a 200) a API de bots do Telegram.\n","\n","import os\n","import json\n","import logging\n","from datetime import datetime, timezone, timedelta\n","\n","import boto3\n","\n","\n","def lambda_handler(event: dict, context: dict) -> dict:\n","\n","  '''\n","  Recebe uma mensagens do Telegram via AWS API Gateway, verifica no\n","  seu conteúdo se foi produzida em um determinado grupo e a escreve, \n","  em seu formato original JSON, em um bucket do AWS S3.\n","  '''\n","\n","  # vars de ambiente\n","\n","  BUCKET = os.environ['AWS_S3_BUCKET']\n","  TELEGRAM_CHAT_ID = int(os.environ['TELEGRAM_CHAT_ID'])\n","\n","  # vars lógicas\n","\n","  tzinfo = timezone(offset=timedelta(hours=-3))\n","  date = datetime.now(tzinfo).strftime('%Y-%m-%d')\n","  timestamp = datetime.now(tzinfo).strftime('%Y%m%d%H%M%S%f')\n","\n","  filename = f'{timestamp}.json'\n","\n","  # código principal\n","\n","  client = boto3.client('s3')\n","\n","  try:\n","\n","    message = json.loads(event[\"body\"])\n","    chat_id = message[\"message\"][\"chat\"][\"id\"]\n","\n","    if chat_id == TELEGRAM_CHAT_ID:\n","\n","      with open(f\"/tmp/{filename}\", mode='w', encoding='utf8') as fp:\n","        json.dump(message, fp)\n","\n","      client.upload_file(f'/tmp/{filename}', BUCKET, f'telegram/context_date={date}/{filename}')\n","\n","  except Exception as exc:\n","      logging.error(msg=exc)\n","      return dict(statusCode=\"500\")\n","\n","  else:\n","      return dict(statusCode=\"200\")\n","```\n","\n","**1.3.** `AWS API Gateway` **API REST** com sufixo `-api` para acionar a função `AWS Lambda` criada para receber e armazenar dados crus do `Telegram bot API` no `AWS S3 bucket` sufixo `-raw`;\n","\n","1.3.1. Criar método tipo \"POST\", integração \"Função do Lambda\", \"Integração do proxy do Lambda\" ativa, escolher a função do `AWS Lambda` criada;\n","\n","1.3.2. Implantar API, Estágio \"Novo estágio\", Nome do estágio \"dev\", guardar \"Invocar URL\" em \"aws_api_gateway_url\" usando a célula de código Python abaixo;\n","\n","> Nota: não disponibilize o endereço da API gerada.\n","\n","**1.4.** `setWebhook` via `Telegram bot API` para redirecionar as informações de mensagens captadas pelo bot para o endereço \"aws_api_gateway_url\" usando a célula de código Python abaixo;\n","\n","> Nota: não disponibilize o token da API de bots do Telegram."]},{"cell_type":"code","execution_count":4,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:35:17.740394Z","iopub.status.busy":"2024-06-09T23:35:17.739987Z","iopub.status.idle":"2024-06-09T23:35:20.654150Z","shell.execute_reply":"2024-06-09T23:35:20.652934Z","shell.execute_reply.started":"2024-06-09T23:35:17.740363Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["Insert AWS API Gateway Invoke URL\n"]},{"name":"stdout","output_type":"stream","text":[" ··························································\n"]},{"name":"stdout","output_type":"stream","text":["aws_api_gateway_url stored\n"]}],"source":["### guarda \"Invocar URL\" em \"aws_api_gateway_url\"\n","\n","from getpass import getpass\n","\n","print(\"Insert AWS API Gateway Invoke URL\")\n","aws_api_gateway_url = getpass()\n","print(\"aws_api_gateway_url stored\")"]},{"cell_type":"code","execution_count":9,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:36:52.994093Z","iopub.status.busy":"2024-06-09T23:36:52.993717Z","iopub.status.idle":"2024-06-09T23:36:53.759316Z","shell.execute_reply":"2024-06-09T23:36:53.757968Z","shell.execute_reply.started":"2024-06-09T23:36:52.994062Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["{\n","  \"ok\": true,\n","  \"result\": true,\n","  \"description\": \"Webhook was set\"\n","}\n"]}],"source":["### configura o webhook para redirecionar as mensagens para a url do AWS API Gateway\n","\n","import json\n","import requests\n","\n","response = requests.get(url=f'{base_url}/setWebhook?url={aws_api_gateway_url}')\n","print(json.dumps(json.loads(response.text), indent=2))"]},{"cell_type":"code","execution_count":null,"metadata":{"trusted":true},"outputs":[],"source":["# ### este trecho apaga e redefine o Webhook\n","# import json\n","# import requests\n","\n","# requests.get(url=f'{base_url}/deleteWebhook')\n","# print(\"Webhook reseted\")"]},{"cell_type":"code","execution_count":10,"metadata":{"execution":{"iopub.execute_input":"2024-06-09T23:36:58.433157Z","iopub.status.busy":"2024-06-09T23:36:58.432775Z","iopub.status.idle":"2024-06-09T23:36:59.189671Z","shell.execute_reply":"2024-06-09T23:36:59.188348Z","shell.execute_reply.started":"2024-06-09T23:36:58.433122Z"},"trusted":true},"outputs":[{"name":"stdout","output_type":"stream","text":["{\n","  \"ok\": true,\n","  \"result\": {\n","    \"url\": \"https://hppmtvbty7.execute-api.sa-east-1.amazonaws.com/dev\",\n","    \"has_custom_certificate\": false,\n","    \"pending_update_count\": 0,\n","    \"max_connections\": 40,\n","    \"ip_address\": \"54.233.185.14\"\n","  }\n","}\n"]}],"source":["### getWebhookInfo retorna as informações sobre o webhook configurado\n","\n","import json\n","import requests\n","\n","response = requests.get(url=f'{base_url}/getWebhookInfo')\n","print(json.dumps(json.loads(response.text), indent=2))"]},{"cell_type":"markdown","metadata":{},"source":["## 2/3\\. ETL\n","\n","**2.1.** `AWS S3 bucket` com o sufixo `-enriched` para armazenar **dados enriquecidos**;\n","\n","**2.2.** `AWS Lambda` **função Python** que transforma o JSON do dia anterior (D-1) do bucket sufixo `-raw` e salva em um arquivo PARQUET particionado por dia no bucket sufixo `-enriched`;\n","\n","2.2.1. variáveis de ambiente para **nome do bucket de dados crus** (AWS_S3_BUCKET) e **nome do bucket de dados enriquecidos** (AWS_S3_ENRICHED) em \"Configuração > Variáveis de Ambiente > Editar\" no painel da função;\n","\n","2.2.2. permissão para escrita no bucket de dados enriquecidos em \"Configuração > Permissôes > Nome da Função [<Função>]\" no painel da função e no `AWS IAM` \"Adicionar permissões > Anexar políticas\" para o `AWS S3 bucket` sufixo `-enriched`;\n","\n","2.2.3. tempo limite para 5 minutos em \"Configuração > Configuração geral > Editar\" no painel da função;\n","\n","2.2.4. camada Python PyArrow em \"Código > Camadas > Adicionar uma camada > criar uma nova camada > Fazer upload de um arquivo do Amazon S3 ([carregar arquivo para Python compatível](https://github.com/awslabs/aws-data-wrangler/releases) em um `AWS S3 bucket` exclusivo) > inserir Link do URL do Amazon S3 > Tempos de execução compatíveis > Camadas personalizadas > Escolher > Versão 1 > Adicionar\" no painel da função;\n","\n","```python\n","### código Python da função `AWS Lambda` criada para dados enriquecidos:\n","\n","# Lista todos os arquivos JSON de uma única participação da camada crua de um bucket do AWS S3;\n","# Para cada arquivo listado:\n","# Faz o download do arquivo e carrega o conteúdo da mensagem;\n","# Executa uma função de data wrangling;\n","# Cria uma tabela do PyArrow e a contatena com as demais.\n","# Persiste a tabela no formato Parquet na camada enriquecida em um bucket do AWS S3\n","\n","import os\n","import json\n","import logging\n","from datetime import datetime, timedelta, timezone\n","\n","import boto3\n","import pyarrow as pa\n","import pyarrow.parquet as pq\n","\n","\n","def lambda_handler(event: dict, context: dict) -> bool:\n","\n","  '''\n","  Diariamente é executado para compactar as diversas mensagensm, no formato\n","  JSON, do dia anterior, armazenadas no bucket de dados cru, em um único \n","  arquivo no formato PARQUET, armazenando-o no bucket de dados enriquecidos\n","  '''\n","\n","  # vars de ambiente\n","\n","  RAW_BUCKET = os.environ['AWS_S3_BUCKET']\n","  ENRICHED_BUCKET = os.environ['AWS_S3_ENRICHED']\n","\n","  # vars lógicas\n","\n","  tzinfo = timezone(offset=timedelta(hours=-3))\n","  date = (datetime.now(tzinfo) - timedelta(days=1)).strftime('%Y-%m-%d')\n","  timestamp = datetime.now(tzinfo).strftime('%Y%m%d%H%M%S%f')\n","\n","  # código principal\n","\n","  table = None\n","  client = boto3.client('s3')\n","\n","  try:\n","\n","      response = client.list_objects_v2(Bucket=RAW_BUCKET, Prefix=f'telegram/context_date={date}')\n","\n","      for content in response['Contents']:\n","\n","        key = content['Key']\n","        client.download_file(RAW_BUCKET, key, f\"/tmp/{key.split('/')[-1]}\")\n","\n","        with open(f\"/tmp/{key.split('/')[-1]}\", mode='r', encoding='utf8') as fp:\n","\n","          data = json.load(fp)\n","          data = data[\"message\"]\n","\n","        parsed_data = parse_data(data=data)\n","        iter_table = pa.Table.from_pydict(mapping=parsed_data)\n","\n","        if table:\n","\n","          table = pa.concat_tables([table, iter_table])\n","\n","        else:\n","\n","          table = iter_table\n","          iter_table = None\n","          \n","      pq.write_table(table=table, where=f'/tmp/{timestamp}.parquet')\n","      client.upload_file(f\"/tmp/{timestamp}.parquet\", ENRICHED_BUCKET, f\"telegram/context_date={date}/{timestamp}.parquet\")\n","\n","      return True\n","  \n","  except Exception as exc:\n","      logging.error(msg=exc)\n","      return False\n","    \n","\n","# data wrangling\n","\n","def parse_data(data: dict) -> dict:\n","\n","  date = datetime.now().strftime('%Y-%m-%d')\n","  timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')\n","\n","  parsed_data = dict()\n","\n","  for key, value in data.items():\n","\n","      if key == 'from':\n","          for k, v in data[key].items():\n","              if k in ['id', 'is_bot', 'first_name']:\n","                parsed_data[f\"{key if key == 'chat' else 'user'}_{k}\"] = [v]\n","\n","      elif key == 'chat':\n","          for k, v in data[key].items():\n","              if k in ['id', 'type']:\n","                parsed_data[f\"{key if key == 'chat' else 'user'}_{k}\"] = [v]\n","\n","      elif key in ['message_id', 'date', 'text']:\n","          parsed_data[key] = [value]\n","\n","  if not 'text' in parsed_data.keys():\n","    parsed_data['text'] = [None]\n","\n","  return parsed_data\n","```\n","\n","**2.3.** `AWS EventBridge` para acionar a função `AWS Lambda` criada para transformar o JSON do dia anterior e armazenar o PARQUET no `AWS S3 bucket` sufixo `-enriched` todos os dias 00h (GMT-3) em \"Criar regra > Nome, Programação > Continuar a criar regra > cron(0 3 * * ? *) > Próximo > Serviço AWS, Função do Lambda, Função (sufixo `-enriched`) > Próximo (x2) > Criar regra\";"]},{"cell_type":"markdown","metadata":{},"source":["## 3/3\\. Apresentação\n","\n","**3.1.** Tabela com os dados enriquecidos utilizando SQL no `AWS Athena`;\n","\n","```sql\n","# Consulta SQL para criar tabela com os dados enriqeucidos particionados por dia:\n","\n","CREATE EXTERNAL TABLE `telegram`(\n","  `message_id` bigint, \n","  `user_id` bigint, \n","  `user_is_bot` boolean, \n","  `user_first_name` string, \n","  `chat_id` bigint, \n","  `chat_type` string, \n","  `text` string, \n","  `date` bigint)\n","PARTITIONED BY ( \n","  `context_date` date)\n","ROW FORMAT SERDE \n","  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' \n","STORED AS INPUTFORMAT \n","  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' \n","OUTPUTFORMAT \n","  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'\n","LOCATION\n","  's3://<bucket-enriquecido>/<pasta-particionada>'\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-create-table.png?raw=true)\n","\n","**3.2.** Carrega as partições na tabela criada;\n","\n","```sql\n","MSCK REPAIR TABLE `telegram`;\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-repair-table.png?raw=true)\n","\n","3.2.1. Para atualizar a tabela também é possível utilizar o comando a seguir para reduzir possíveis custos de consulta;\n"," ```sql\n"," ALTER TABLE <nome-tabela> ADD PARTITION <coluna-partição> = <valor-partição>\n","  ```\n","\n","**3.3.** Dados analíticos;\n","\n","- Quantidade de mensagens por dia;\n","\n","```sql\n","SELECT \n","  context_date, \n","  count(1) AS \"message_amount\" \n","FROM \"telegram\" \n","GROUP BY context_date \n","ORDER BY context_date DESC\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-select-message_amount.png?raw=true)\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/graph-select-message_amount.png?raw=true)\n","\n","- Quantidade de mensagens por usuário por dia.\n","\n","```sql\n","SELECT \n","  user_id, \n","  user_first_name, \n","  context_date, \n","  count(1) AS \"message_amount\" \n","FROM \"telegram\" \n","GROUP BY \n","  user_id, \n","  user_first_name, \n","  context_date \n","ORDER BY context_date DESC\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-select-user-message_amount.png?raw=true)\n","\n","- Média do tamanho das mensagens por usuário por dia.\n","\n","```sql\n","SELECT \n","  user_id, \n","  user_first_name, \n","  context_date,\n","  CAST(AVG(length(text)) AS INT) AS \"average_message_length\" \n","FROM \"telegram\" \n","GROUP BY \n","  user_id, \n","  user_first_name, \n","  context_date \n","ORDER BY context_date DESC\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-select-user-message_lenght.png?raw=true)\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/graph-select-user-message_lenght.png?raw=true)\n","\n","- Quantidade de mensagens por hora por dia da semana por número da semana.\n","\n","```sql\n","WITH \n","parsed_date_cte AS (\n","    SELECT \n","        *, \n","        CAST(date_format(from_unixtime(\"date\"),'%Y-%m-%d %H:%i:%s') AS timestamp) AS parsed_date\n","    FROM \"telegram\" \n","),\n","hour_week_cte AS (\n","    SELECT\n","        *,\n","        EXTRACT(hour FROM parsed_date) AS parsed_date_hour,\n","        EXTRACT(dow FROM parsed_date) AS parsed_date_weekday,\n","        EXTRACT(week FROM parsed_date) AS parsed_date_weeknum\n","    FROM parsed_date_cte\n",")\n","SELECT\n","    parsed_date_hour,\n","    parsed_date_weekday,\n","    parsed_date_weeknum,\n","    count(1) AS \"message_amount\" \n","FROM hour_week_cte\n","GROUP BY\n","    parsed_date_hour,\n","    parsed_date_weekday,\n","    parsed_date_weeknum\n","ORDER BY\n","    parsed_date_weeknum,\n","    parsed_date_weekday\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-select-hour-week-message_amount.png?raw=true)\n","\n","-  Total de vezes que cada palavra única aparece em todas as mensagens\n","\n","```sql\n","WITH words AS (\n","  SELECT word\n","  FROM \"telegram\",\n","  UNNEST(split(\"text\", ' ')) AS t(word)\n",")\n","SELECT word, COUNT(*) as count\n","FROM words\n","GROUP BY word\n","ORDER BY count DESC;\n","```\n","\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/athena-select-count-unnest-words.png?raw=true)\n","![](https://github.com/rafie-b/Data-Warehouse-AWS-Pipeline-Chat-API/blob/main/repo-Data-Warehouse-AWS-Pipeline-Chat-API/graph-select-count-unnest-words.png?raw=true)"]}],"metadata":{"kaggle":{"accelerator":"none","dataSources":[],"dockerImageVersionId":30673,"isGpuEnabled":false,"isInternetEnabled":true,"language":"python","sourceType":"notebook"},"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.10.13"}},"nbformat":4,"nbformat_minor":4}