小编Ale*_*dro的帖子

我有这张桌子:

CREATE TABLE `ClientesHora_copy` (
`dia` varchar(6) default NULL,
`hora` varchar(2) default NULL,
`sit` varchar(17) default NULL,
`nodo` varchar(6) default NULL,
`clientes` decimal(41,0) default NULL,
`segundos` double default NULL,
`llamadas` decimal(41,0) default NULL,
`fecha` datetime default NULL,
KEY `nodo_fecha` (`nodo`,`fecha`),
KEY `nodo` (`nodo`)
) ENGINE=MyISAM DEFAULT CHARSET=latin1

而这个查询:

SET @sitio= 'ABA000';
SET @horaini='2013-02-12 15:18:00';
SET @horafin='2013-02-12 20:36:00';     
    EXPLAIN SELECT nodo,sit,clientes,segundos,llamadas,fecha,hora,@horaini AS  horaini,@horafin AS horafin
        FROM `ClientesHora_copy`
        WHERE 
        nodo =@sitio
        AND
        fecha BETWEEN DATE_SUB(DATE_FORMAT(@horaini, "%Y-%m-%d %H:00:00"), INTERVAL 7 DAY)
            AND DATE_SUB(DATE_FORMAT(@horafin, …

我正在尝试左连接 Spark 3 上的 2 个表，其中包含 17M 行（事件）和 400M 行（详细信息）。拥有一个包含 1 + 15 x 64 核实例的 EMR 集群。（r6g.16xlarge 尝试使用类似的 r5a）源文件是从 S3 加载的未分区镶木地板。

这是我用来加入的代码：

join = (
    broadcast(events).join(
        details,
        [
            details["a"] == events["a2"],
            (unix_timestamp(events["date"]) - unix_timestamp(details["date"])) / 3600
            > 5,
        ],
        "left",
    )
).drop("a")

join.checkpoint()

为了分区，我使用这个：

executors = 15 * 64 * 3  # 15 instances, 64 cores, 3 workers per core

所以我尝试了：

details = details.repartition(executors, "a")

和

details = details.withColumn("salt", (rand(seed=42) * nSaltBins).cast("int"))
details = details.repartition(executors, "salt")

在这两种情况下，90％的工作人员在大约 5-10 分钟内结束，其余的继续很长一段时间（50 …

我有两个通过 pyodbc 的 Mysql 连接声明为：

connMy1 = pyodbc.connect('DRIVER={MySQL ODBC 5.3 Unicode Driver};SERVER=***;UID=***;PWD=***')
connMy1.autocommit = True
cursorMy1 = connMy1.cursor()

connMy2 = pyodbc.connect('DRIVER={MySQL ODBC 5.3 Unicode Driver};SERVER=***;UID=***;PWD=***')
connMy2.autocommit = True
cursorMy2 = connMy2.cursor()

我创建了一个带有熊猫的 CSV 连接，如下所示：

def bajar(sql,tabla,ruta):
    print ("bajando datos")
    chunk = 10 ** 5 
    chunks = pandas.read_sql(sql, connMy1, chunksize=chunk)
    eliminarArchivo(ruta)
    print ("creando CSV")
    with open(ruta, 'w') as output:
        for n, df in enumerate(chunks):
            write_header = n == 0
            df.to_csv(output, sep=';', index=False, header=False, na_rep='NULL')
    connMy1.commit()   

然后我调用这个函数来上传 CSV

def subir(ruta,tabla):
    print ("Subiendo …

我正在尝试从 Netezza 导出一个大文件（使用 Netezza ODBC + pyodbc），此解决方案会引发 memoryError，如果我在没有“list”的情况下循环，它会非常慢。你知道一个不会杀死我的服务器/python进程但可以运行得更快的中间解决方案吗？

cursorNZ.execute(sql)
archi = open("c:\test.csv", "w")
lista = list(cursorNZ.fetchall())
for fila  in lista:
    registro = ''
    for campo in fila:
        campo = str(campo)
        registro = registro+str(campo)+";"
    registro = registro[:-1]
    registro = registro.replace('None','NULL')
    registro = registro.replace("'NULL'","NULL")
    archi.write(registro+"\n")

- - 编辑 - -

谢谢，我正在尝试：其中“sql”是查询，cursorNZ 是

connMy = pyodbc.connect(DRIVER=.....)
cursorNZ = connNZ.cursor()

chunk = 10 ** 5  # tweak this
chunks = pandas.read_sql(sql, cursorNZ, chunksize=chunk)
with open('C:/test.csv', 'a') as output:
    for n, df in enumerate(chunks):
        write_header …