标签: etl

我已经在谷歌上搜索了一个教程来帮助解决这个问题，但还没有找到任何全面的内容。

我想通过以 JSON 格式发送数据库中包含的数据来将 SQLite3 数据库与 Web 服务进行单向同步，但无法找到有关如何将数据库转换为 JSON 的信息。如果有人可以向我指出涵盖此内容的教程的方向，或者展示一个如何转换简单 SQLite 表的简短示例，那就太好了！

该数据库包含大约十个表，可能包含大量数据，因此除了简单的示例之外，我们将不胜感激一些一般性指导。

谢谢！

我正在用 Pentaho 的勺子设计一个 ETL。

我的目的是在“加载到数据库”和“更新数据库”之后执行“计算值”。

应该用什么步骤来替换 '??? 等等'？

在这篇文章中，我不是在问任何教程，如何做某事，在这篇文章中，我请求您的帮助，如果有人可以用简单的话解释我，什么是 DWH（数据仓库）和什么是 ETL。

当然，我在谷歌和 YouTube 上搜索了很多，我发现了很多文章、视频，但我仍然不太确定它是什么。

我为什么要问？
在申请工作之前我需要非常了解它。

我在这里看到了维度查找/更新文档和其他一些博客。但我似乎无法得到一个清晰的想法。我有一个具有以下结构的表：

 Key Name Code Status IN Out Active

来自key name code status activecsv 文件。我需要对 scd type2 使用维度查找/更新步骤并填充IN/Out.

设置连接详细信息后，我已KEY 使用选项将键设置为并将字段设置为所有其他字段Date of last insert (without stream field as source)。如果任何其他字段发生更改，我需要在数据库中创建一个新行。该行将具有相同key且更新的详细信息，设置out为无穷大并in设置为当前系统日期。

日期范围开始字段设置为IN，表日期范围结束字段设置为OUT数据库列。

我不明白这个技术密钥的概念，因为密钥也来自 csv 文件。

当我点击预览时出现错误：

   DB2 SQL error: SQLCODE: -407, SQLSTATE: 23502, SQLERRMC: 

请让我知道更多详细信息以及我错过的任何步骤或设置。

我正在尝试使用 Python 将大量 JSON 文件从 Amazon S3 导入到 AWS RDS-PostgreSQL 中。但是，这些错误发生了，

回溯（最近一次调用最后一次）：

文件“my_code.py”，第 67 行，位于

file_content = obj['Body'].read().decode('utf-8').splitlines(True)

文件“/home/user/asd-to-qwe/fgh-to-hjk/env/local/lib/python3.6/site-packages/botocore/response.py”，第 76 行，读取

块= self._raw_stream.read(amt)

文件“/home/user/asd-to-qwe/fgh-to-hjk/env/local/lib/python3.6/site-packages/botocore/vendored/requests/packages/urllib3/response.py”，第239行，在读

数据 = self._fp.read()

文件“/usr/lib64/python3.6/http/client.py”，第 462 行，读取

s = self._safe_read(self.length)

文件“/usr/lib64/python3.6/http/client.py”，第 617 行，在 _safe_read 中

返回 b"".join(s)

内存错误

// my_code.py

import sys
import boto3
import psycopg2
import zipfile
import io
import json

s3 = boto3.client('s3', aws_access_key_id=<aws_access_key_id>, aws_secret_access_key=<aws_secret_access_key>)
connection = psycopg2.connect(host=<host>, dbname=<dbname>, user=<user>, password=<password>)
cursor = connection.cursor()

bucket = sys.argv[1]
key = sys.argv[2]
obj = s3.get_object(Bucket=bucket, Key=key)

def …

我想自动化每天运行的 redshift 插入查询。

我们实际上使用的是Aws环境。有人告诉我使用 lambda 不是正确的方法。这是在 Redshift 中自动执行查询的最佳 ETL 流程。

我正在尝试使用气流按计划使用气流在 Athena 中运行查询。

我在下面包含的函数是使用PythonOperatorin运行的airflow。

from airflow.models import Variable
from airflow.contrib.hooks.aws_athena_hook import AWSAthenaHook
import datetime

def update_athena_partition(*args, **kwargs):
    execution_date = datetime.datetime.strptime(kwargs['ds'], '%Y-%m-%d')
    execution_month = execution_date.month
    execution_year = execution_date.year
    s3_prefix = Variable.get('bikeshare_s3_prefix')
    bucket_name = Variable.get('bikeshare_bucket_name')
    athena_table_name = Variable.get('bikeshare_athena_table')
    result_configuration = {"OutputLocation": "s3://{}/".format(bucket_name)}
    file_location = 's3://bucket_name/' + s3_prefix + f'year=2018/month=2/'
    partition_update_query = """
    ALTER TABLE {} add partition (year="{}", month='{}')
    location "{}";
    """
    athena_hook = AWSAthenaHook(aws_conn_id='aws_credentials')
    athena_hook.run_query(partition_update_query.format(athena_table_name,
                                                        2018,
                                                        2,
                                                        file_location),
                          result_configuration=result_configuration,
                          query_context="athena_database_name")

这是我的 DAG

etl_dag = …

我有一个将 CSV 文件导入到数据库的 SSIS 包，并且该包将每天运行，因为它将新行附加到表的底部并且datetime目标表中没有列。

我想要做的是在目标表中创建一个新的date/timestamp并且每次加载数据时我都希望该表反映数据库中的加载时间Timestamp。

我不确定我是如何实现这一目标的。

我想使用Aggregate.

这是场景：

我有一张表，其值如下：

Key | Value
1   | v1
1   | v2
2   | v1
2   | v3

执行聚合后，我希望输出如下：

Key | Value
1   | v1, v2
2   | v1, v3

我试图找到String_AggADF 中没有的功能。

我正在尝试在Excel文件输入的某些列之间进行聚合操作.我有以下情况:

Column 1   Column 2    Column 3

X            $15           A
X            $20           A
Y            $1        B
Y            $1        B
Y            $3        C

我想实现这种聚合操作:

Column 1    Column 2    Column 3

X             $35           A
Y             $2            B
Y             $3            C

如您所见,第1列和第3列是执行聚合操作的标准,在这种情况下,我想获得第2列的总和.

在Pentaho数据集成中有没有办法做到这一点？我试过"加入行"和"加入行(作为笛卡尔产品)",但是,我没有结果.