Імпорт великих dbf файлів в Ms SQL server 2008 з допомогою SqlBulkCopy

У цій статті я розповім як завантажити багато величезних dbf файлів, що складаються з мільйонів записів у вашу базу даних на ms sql сервері за прийнятний час.

Завдання на перший погляд тривіальна. Можна використовувати майстер в sql management studio або функцію OPENROWSET через запит.

Але перший варіант після декількох спроб відпав через різних глюків і необхідності завантаження великої кількості файлів в одну таблицю (близько 100 файлів). До то чоловіка при тривалій завантаженні вилітала помилка.

Другий варіант теж не підійшов за різної розрядності драйверів і розрядності сервера.

Так як файл просто величезний, то було вирішено читати його через потік і записуються в базу. Далі після прочитання рядка у файлі треба цю рядок записати в таблицю. Перше що спало на думку це використовувати insert, але запис у цьому разі зайняла б надто багато часу.

І тут я згадав про інший механізм запису через SqlBulkCopy, який дозволяє заливати величезне число записів без запитів insert.
На ділі це використання класу SqlBulkCopy, для здійснення запису через який треба реалізувати лише один інтерфейс IDataReader.

Отже почнемо з реалізації інтерфейсу
public class BDFBulkReader : IDataReader


Почнемо з функції, яка повертає значення поточного запису:
public object GetValue(int i) { return R[FieldIndex[i]]; }

Зверну вашу увагу на те, що поля в файлі і поля в таблиці можуть бути в різному порядку. А за індексом хотілося б отримувати значення для відповідного поля таблиці. Тому я використовував додатково словник FieldIndex, де зіставлення імен полів номером в таблиці sql. За номером береться ім'я поля, на ім'я зі словника R береться значення прочитаного рядка dbf файлу. У результаті для n-го індексу в бд GetValue поверне відповідне значення.
Dictionary<string, object> R = new Dictionary<string, object>();
Словник<int, string> FieldIndex = new Dictionary<int, string>();


FieldIndex будемо передавати вже заповнений для таблиці, а R буде заповнювати при виклику самими reader'ом функції Read, яку в подальшому теж реалізуємо.

Отже, конструктор:

System.IO.FileStream FS;
byte[] buffer;
int _FieldCount;
int FieldsLength;
System.Globalization.DateTimeFormatInfo dfi = new System.Globalization.CultureInfo("en-US", false).DateTimeFormat;
System.Globalization.NumberFormatInfo nfi = new System.Globalization.CultureInfo("en-US", false).NumberFormat;
string[] FieldName;
string[] FieldType;
byte[] FieldSize;
byte[] FieldDigs;
int RowsCount;
int ReadedRow = 0;
Словник<string, object> R = new Dictionary<string, object>();
Словник<int, string> FieldIndex = new Dictionary<int, string>();

public BDFBulkReader(string FileName Dictionary<int, string> FieldIndex)
{
FS = new System.IO.FileStream(FileName, System.IO.FileMode.Open);
buffer = new byte[4]; 
FS.Position = 4; FS.Read(buffer, 0, buffer.Length);
RowsCount = buffer[0] + (buffer[1] * 0x100) + (buffer[2] * 0x10000) + (buffer[3] * 0x1000000);
buffer = new byte[2]; 
FS.Position = 8; FS.Read(buffer, 0, buffer.Length);
_FieldCount = (((buffer[0] + (buffer[1] * 0x100)) - 1) / 32) - 1;
FieldName = new string[_FieldCount]; 
FieldType = new string[_FieldCount]; 
FieldSize = new byte[_FieldCount]; 
FieldDigs = new byte[_FieldCount];
buffer = new byte[32 * _FieldCount];
FS.Position = 32; FS.Read(buffer, 0, buffer.Length);
FieldsLength = 0;
for (int i = 0; i < _FieldCount; i++)
{ 
FieldName[i] = System.Text.Encoding.Default.GetString(buffer, i * 32, 10).TrimEnd(new char[] { (char)0x00 });
FieldType[i] = "" + (char)buffer[i * 32 + 11];
FieldSize[i] = buffer[i * 32 + 16];
FieldDigs[i] = buffer[i * 32 + 17];
FieldsLength = FieldsLength + FieldSize[i];
}
FS.ReadByte(); 

this.FieldIndex = FieldIndex;
}


Його завдання відкрити файл, визначити імена полів, їх кількість та їх типи. Другий параметр конструктора, як я писав вище, це словник відповідностей, щоб, наприклад, за першим номером поля ми гарантовано отримали потрібне поле з файлу.

Тепер перейдемо до реалізації bool Read(). Вона поверне true у разі якщо рядок успішно прочитано. І false у випадку якщо рядок не була прочитана і в той же час був досягнутий кінець даних.

public bool Read()
{
if (ReadedRow >= RowsCount) return false;

R. Clear();
buffer = new byte[FieldsLength];
FS.ReadByte(); 
FS.Read(buffer, 0, buffer.Length);
int Index = 0;
for (int i = 0; i < FieldCount; i++)
{
string l = System.Text.Encoding.GetEncoding(866).GetString(buffer, Index, FieldSize[i]).TrimEnd(new char[] { (char)0x00 }).TrimEnd(new char[] { (char)0x20 });
Index = Index + FieldSize[i];
object Tr;
if (l.Trim() != "")
{
switch (FieldType[i])
{
case "L": Tr = l == "T" ? true : false; break;
case "D": Tr = DateTime.ParseExact(l, "yyyyMMdd", dfi); break;
case "N":
{
if (FieldDigs[i] == 0)
Tr = int.Parse(l, nfi);
else
Tr = decimal.Parse(l, nfi);
break;
}
case "F": Tr = double.Parse(l, nfi); break;
default: Tr = l; break;
}

}
else
{
Tr = DBNull.Value;
}
R. Add(FieldName[i], Tr);
}
ReadedRow++;
return true;
}


Ще раз нагадаю, що після її виклику прочитана рядок запишеться у словник R, для подальшого читання reader'ом.
Отже, залишилося реалізувати, метод повертає кількість полів:

public int FieldCount { get { return _FieldCount; } }


І заглушки для інтерфейсу:

public void Dispose() { FS.Close(); } 
public int Depth { get { return -1; } }
public bool IsClosed { get { return false; } }
public Object this[int i] { get { return new object(); } }
public Object this[string name] { get { return new object(); } }
public int RecordsAffected { get { return -1; } }

public void Close() { }
public bool NextResult() { return true; }
public bool IsDBNull(int i) { return false; }
public string GetString(int i) { return ""; }
public DataTable GetSchemaTable() { return null; }
public int GetOrdinal(string name) { return -1; }
public string GetName(int i) { return ""; }
public long GetInt64(int i) { return -1; }
public int GetInt32(int i) { return -1; }
public short GetInt16(int i) { return -1; }
public Guid GetGuid(int i) { return new Guid(); }
public float GetFloat(int i) { return -1; }
public Type GetFieldType(int i) { return typeof(string); }
public double GetDouble(int i) { return -1; }
public decimal GetDecimal(int i) { return -1; }
public DateTime GetDateTime(int i) { return new DateTime(); }
public string GetDataTypeName(int i) { return ""; }
public IDataReader GetData(int i) { return this; }
public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length) { return -1; }
public char GetChar(int i) { return ' '; }
public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length) { return -1; }
public byte GetByte(int i) { return 0x00; }
public bool GetBoolean(int i) { return false; }
public int GetValues(Object[] values) { return -1; }


Де в Dispose() я просто закриваю файл.

Після того як інтерфейс реалізований, можна написати метод для завантаження файлу:

void SaveToTable(FileInfo dir, string TableName, string connestionString Dictionary<int, string> FieldIndex)
{
using (var loader = new SqlBulkCopy(connestionString, SqlBulkCopyOptions.Default))
{ 
loader.DestinationTableName = TableName;
loader.BulkCopyTimeout = 9999;
loader.WriteToServer(new BDFBulkReader(dir.FullName, FieldIndex)); 
}
}


От і все. В цю функцію залишилося передати розташування файлу, ім'я таблиці, рядок підключення і відповідний словник відповідностей, наприклад:

Dictionary<int, string> FieldIndex= new Dictionary<int, string>();
FieldIndex.Add(0, "POSTALCODE");
FieldIndex.Add(1, "IFNSFL");
FieldIndex.Add(2, "TERRIFNSFL");
FieldIndex.Add(3, "IFNSUL");
FieldIndex.Add(4, "TERRIFNSUL");
FieldIndex.Add(5, "OKATO");
FieldIndex.Add(6, "OKTMO");
FieldIndex.Add(7, "UPDATEDATE");
FieldIndex.Add(8, "HOUSENUM");
FieldIndex.Add(9, "ESTSTATUS");
FieldIndex.Add(10, "BUILDNUM");
FieldIndex.Add(11, "STRUCNUM");
FieldIndex.Add(12, "STRSTATUS");
FieldIndex.Add(13, "HOUSEID");
FieldIndex.Add(14, "HOUSEGUID");
FieldIndex.Add(15, "AOGUID");
FieldIndex.Add(16, "STARTDATE");
FieldIndex.Add(17, "a list");
FieldIndex.Add(18, "STATSTATUS");
FieldIndex.Add(19, "NORMDOC");
FieldIndex.Add(20, "COUNTER");


Все, всім дякую за увагу, приємною завантаження.

Корисні посилання:
Опис SqlBulkCopy
Нутрощі dbf

Джерело: Хабрахабр

0 коментарів

Тільки зареєстровані та авторизовані користувачі можуть залишати коментарі.